merge: integrate agent-mvp into master

Merge PR #1 after backend security and contract gates passed.

.dockerignore

@@ -11,9 +11,12 @@ dist/
 package/
 temp/
 data/
-# db_inp/
+db_inp/
 inp/
-# .env
+.env
+.env.*
+logs/
+coverage/
 *.pyc
 *.dump
 app/algorithms/health/model/my_survival_forest_model_quxi.joblib

.gitea/workflows/package.yml

@@ -1,263 +1,27 @@
-name: Server CI/CD
+name: Server CI/CD v2
 
 on:
   push:
     tags:
       - "v*"
-      - "latest"
   workflow_dispatch: {}
 
 jobs:
-  docker-image:
-    runs-on: ubuntu-22.04
-    if: startsWith(github.ref, 'refs/tags/')
-    permissions:
-      contents: read
-    defaults:
-      run:
-        shell: bash
-
-    steps:
-      - name: Checkout repository
-        uses: https://gitea.waternetwork.cn/actions/checkout@v4
+  build-test-publish-and-deploy:
+    uses: OrgTJWater/ci-templates/.gitea/workflows/container-cd.yml@main
     with:
-          fetch-depth: 1
-
-      - name: Normalize image metadata
-        env:
-          RAW_REGISTRY_HOST: ${{ vars.REGISTRY_HOST }}
-          RAW_REPOSITORY: ${{ github.repository }}
-          RAW_REF_NAME: ${{ github.ref_name }}
-        run: |
-          RAW_REGISTRY_HOST="$(printf '%s' "${RAW_REGISTRY_HOST}" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
-
-          if [ -z "${RAW_REGISTRY_HOST}" ]; then
-            echo "Missing required repository variable: REGISTRY_HOST"
-            exit 1
-          fi
-
-          REGISTRY_HOST="${RAW_REGISTRY_HOST#http://}"
-          REGISTRY_HOST="${REGISTRY_HOST#https://}"
-          REGISTRY_HOST="${REGISTRY_HOST%/}"
-
-          if [ -z "${REGISTRY_HOST}" ]; then
-            echo "Repository variable REGISTRY_HOST resolves to an empty host"
-            exit 1
-          fi
-
-          REPOSITORY_PATH="${RAW_REPOSITORY#/}"
-          IMAGE_OWNER="${REPOSITORY_PATH%%/*}"
-          IMAGE_REPOSITORY_PATH="$(printf '%s' "${IMAGE_OWNER}/tjwater-backend" | tr '[:upper:]' '[:lower:]')"
-          IMAGE_NAME="${REGISTRY_HOST}/${IMAGE_REPOSITORY_PATH}"
-          IMAGE_TAG="${RAW_REF_NAME}"
-          {
-            echo "REGISTRY_HOST=${REGISTRY_HOST}"
-            echo "REPOSITORY_PATH=${REPOSITORY_PATH}"
-            echo "IMAGE_REPOSITORY_PATH=${IMAGE_REPOSITORY_PATH}"
-            echo "IMAGE_NAME=${IMAGE_NAME}"
-            echo "IMAGE_TAG=${IMAGE_TAG}"
-            echo "IMAGE_REF=${IMAGE_NAME}:${IMAGE_TAG}"
-          } >> "$GITHUB_ENV"
-
-      - name: Login to Gitea Container Registry
-        env:
+      image_name: gitea.waternetwork.cn/orgtjwater/tjwater-backend
+      dockerfile: Dockerfile
+      build_context: .
+      test_command: |
+        test -f app/api/v1/endpoints/access.py
+        grep -Fq 'api_router.include_router(access.router' app/api/v1/router.py
+        grep -Fq '@router.get("/projects"' app/api/v1/endpoints/meta.py
+        grep -Fq '@router.get("/projects/current"' app/api/v1/endpoints/project.py
+        grep -Fq '@router.post("/audit-events"' app/api/v1/endpoints/audit.py
+      deploy_service: backend
+      deploy_host: 192.168.1.114
+    secrets:
       REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
       REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
-          GITEA_SERVER_URL: ${{ github.server_url }}
-        run: |
-          if [ -z "${REGISTRY_HOST:-}" ]; then
-            echo "Missing resolved environment value: REGISTRY_HOST"
-            exit 1
-          fi
-
-          if [ -z "${REGISTRY_USERNAME}" ]; then
-            echo "Missing required repository secret: REGISTRY_USERNAME"
-            exit 1
-          fi
-
-          if [ -z "${REGISTRY_PASSWORD}" ]; then
-            echo "Missing required repository secret: REGISTRY_PASSWORD"
-            exit 1
-          fi
-
-          echo "Registry username: ${REGISTRY_USERNAME}"
-          echo "Image target: ${IMAGE_REF}"
-
-          API_SERVER_URL="${GITEA_SERVER_URL%/}"
-          api_user="$(curl -fsS \
-            -H "Authorization: token ${REGISTRY_PASSWORD}" \
-            "${API_SERVER_URL}/api/v1/user" \
-            | sed -n 's/.*"login"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' \
-            | head -n 1 || true)"
-
-          if [ -n "${api_user}" ]; then
-            echo "Registry token resolves to Gitea user: ${api_user}"
-          else
-            echo "Could not resolve Gitea user from REGISTRY_PASSWORD token; docker login may still use a password or a token without API access."
-          fi
-
-          echo "Logging into registry host: ${REGISTRY_HOST}"
-          echo "${REGISTRY_PASSWORD}" | docker login "$REGISTRY_HOST" \
-            --username "${REGISTRY_USERNAME}" \
-            --password-stdin
-
-      - name: Materialize runtime env file
-        env:
-          TJWATER_SERVER_ENV: ${{ secrets.TJWATER_SERVER_ENV }}
-        run: |
-          if [ -z "${TJWATER_SERVER_ENV}" ]; then
-            echo "Missing required repository secret: TJWATER_SERVER_ENV"
-            echo "Store the backend .env file content as a multiline Gitea repository secret named TJWATER_SERVER_ENV."
-            exit 1
-          fi
-
-          printf '%s\n' "${TJWATER_SERVER_ENV}" > .env
-          chmod 600 .env
-
-          required_env_keys=(
-            ENVIRONMENT
-            NETWORK_NAME
-            DB_NAME
-            DB_HOST
-            DB_PORT
-            DB_USER
-            DB_PASSWORD
-            TIMESCALEDB_DB_NAME
-            TIMESCALEDB_DB_HOST
-            TIMESCALEDB_DB_PORT
-            TIMESCALEDB_DB_USER
-            TIMESCALEDB_DB_PASSWORD
-            METADATA_DB_NAME
-            METADATA_DB_HOST
-            METADATA_DB_PORT
-            METADATA_DB_USER
-            METADATA_DB_PASSWORD
-            DATABASE_ENCRYPTION_KEY
-          )
-
-          missing_keys=()
-          for key in "${required_env_keys[@]}"; do
-            if ! grep -Eq "^[[:space:]]*(export[[:space:]]+)?${key}[[:space:]]*=" .env; then
-              missing_keys+=("$key")
-            fi
-          done
-
-          if [ "${#missing_keys[@]}" -gt 0 ]; then
-            echo "TJWATER_SERVER_ENV is missing required keys: ${missing_keys[*]}"
-            exit 1
-          fi
-
-      - name: Validate workspace
-        run: |
-          if [ ! -f ./Dockerfile ]; then
-            echo "Dockerfile not found in workspace. Repository checkout may have failed or produced an unexpected workspace."
-            exit 1
-          fi
-
-      - name: Build and Push Image
-        run: |
-          if [ -z "${IMAGE_NAME:-}" ] || [ -z "${IMAGE_TAG:-}" ]; then
-            echo "Missing resolved image metadata: IMAGE_NAME or IMAGE_TAG"
-            exit 1
-          fi
-
-          push_with_retry() {
-            image_ref="$1"
-            attempt=1
-            max_attempts=3
-
-            while [ "$attempt" -le "$max_attempts" ]; do
-              if docker push "$image_ref"; then
-                return 0
-              fi
-
-              if [ "$attempt" -eq "$max_attempts" ]; then
-                return 1
-              fi
-
-              echo "Push failed for $image_ref (attempt $attempt/$max_attempts); retrying in 10s..."
-              attempt=$((attempt + 1))
-              sleep 10
-            done
-          }
-
-          if [ "${IMAGE_TAG}" = "latest" ]; then
-            docker build \
-              -f ./Dockerfile \
-              -t "${IMAGE_NAME}:latest" \
-              .
-            push_with_retry "${IMAGE_NAME}:latest"
-          else
-            docker build \
-              -f ./Dockerfile \
-              -t "${IMAGE_NAME}:${IMAGE_TAG}" \
-              -t "${IMAGE_NAME}:latest" \
-              .
-            push_with_retry "${IMAGE_NAME}:${IMAGE_TAG}"
-            push_with_retry "${IMAGE_NAME}:latest"
-          fi
-
-      - name: Notify Deploy Server
-        run: |
-          post_deploy_webhook() {
-            label="$1"
-            payload="$2"
-            webhook_url="${{ vars.DEPLOY_WEBHOOK_URL }}"
-            token="${{ secrets.DEPLOY_WEBHOOK_TOKEN }}"
-
-            webhook_url=$(echo "$webhook_url" | xargs)
-
-            if [ -z "$webhook_url" ]; then
-              echo "Missing required repository variable: DEPLOY_WEBHOOK_URL"
-              return 1
-            fi
-
-            if [ -z "$token" ]; then
-              echo "Missing required repository secret: DEPLOY_WEBHOOK_TOKEN"
-              return 1
-            fi
-
-            echo "[$label] Calling webhook: $webhook_url"
-
-            http_code=$(curl -sS -D /tmp/deploy_headers.txt -o /tmp/deploy_response.txt -w "%{http_code}" -X POST "$webhook_url" \
-              -H "Content-Type: application/json" \
-              -H "Authorization: Bearer $token" \
-              -d "$payload")
-
-            echo "[$label] webhook HTTP status: ${http_code}"
-            if [ "$http_code" -ge 200 ] && [ "$http_code" -lt 300 ]; then
-              return 0
-            fi
-
-            echo "[$label] response headers:"
-            cat /tmp/deploy_headers.txt
-            echo "[$label] response body:"
-            cat /tmp/deploy_response.txt
-            return 1
-          }
-
-          PRIMARY_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${REPOSITORY_PATH}\"}"
-          FALLBACK_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${IMAGE_REPOSITORY_PATH}\"}"
-
-          echo "Deploy webhook target: ${{ vars.DEPLOY_WEBHOOK_URL }}"
-          echo "Deploy payload(primary): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${REPOSITORY_PATH}"
-          if post_deploy_webhook "primary" "$PRIMARY_PAYLOAD"; then
-            exit 0
-          fi
-
-          echo "Primary webhook request failed, retrying with lowercase repo path..."
-          echo "Deploy payload(fallback): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${IMAGE_REPOSITORY_PATH}"
-          if post_deploy_webhook "fallback" "$FALLBACK_PAYLOAD"; then
-            exit 0
-          fi
-
-          echo "Deploy webhook failed after primary and fallback attempts."
-          exit 1
-
-  deploy-fallback-log:
-    runs-on: ubuntu-22.04
-    needs: docker-image
-    if: failure()
-    steps:
-      - name: Deployment not triggered
-        run: echo "Image build/push failed, deployment webhook was not called."
+      DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}

Dockerfile

@@ -14,13 +14,11 @@ COPY requirements.txt .
 RUN pip install --no-cache-dir uv
 RUN uv pip install --system --no-cache-dir -r requirements.txt
 
-# 将代码放入子目录 'app'，将数据放入子目录 'db_inp'
-# 这样临时文件默认会生成在 /app 下，而代码在 /app/app 下，实现了分离
+# 本地数据目录和环境变量在运行时通过 Compose 挂载或注入，
+# 不应进入镜像构建上下文。
 COPY app ./app
 RUN python -c "from pathlib import Path; from zipfile import ZipFile; model_dir = Path('app/algorithms/health/model'); zip_path = model_dir / 'my_survival_forest_model_quxi.zip'; joblib_name = 'my_survival_forest_model_quxi.joblib'; joblib_path = model_dir / joblib_name; assert zip_path.exists(), f'Model archive not found: {zip_path}'; archive = ZipFile(zip_path); archive.extract(joblib_name, model_dir); archive.close(); assert joblib_path.exists(), f'Model file not extracted: {joblib_path}'" && \
     rm -f app/algorithms/health/model/my_survival_forest_model_quxi.zip
-# COPY db_inp ./db_inp
-COPY .env .
 RUN mkdir -p db_inp temp data inp
 
 # 设置 PYTHONPATH 以便 uvicorn 找到 app 模块

app/algorithms/simulation/scenarios.py

@@ -316,6 +316,7 @@ def flushing_analysis(
     flushing_flow: float = 0,
     scheme_name: str = None,
     username: str | None = None,
+    valve_control: dict[str, dict] = None,
 ) -> None:
     """
     管道冲洗模拟
@@ -323,6 +324,7 @@ def flushing_analysis(
     :param modify_pattern_start_time: 模拟开始时间,格式为'2024-11-25T09:00:00+08:00'
     :param modify_total_duration: 模拟总历时，秒
     :param modify_valve_opening: dict中包含多个阀门开启度，str为阀门的id，float为修改后的阀门开启度
+    :param valve_control: dict中可分别指定阀门的status、setting和k
     :param drainage_node_ID: 冲洗排放口所在节点ID
     :param flushing_flow: 冲洗水量，传入参数单位为m3/h
     :param scheme_name: 方案名称
@@ -334,6 +336,7 @@ def flushing_analysis(
     scheme_detail: dict = {
         "duration": modify_total_duration,
         "valve_opening": modify_valve_opening,
+        "valve_control": valve_control,
         "drainage_node_ID": drainage_node_ID,
         "flushing_flow": flushing_flow,
     }
@@ -450,6 +453,7 @@ def flushing_analysis(
         modify_pattern_start_time=modify_pattern_start_time,
         modify_total_duration=modify_total_duration,
         modify_valve_opening=modify_valve_opening,
+        valve_control=valve_control,
         scheme_type="flushing_analysis",
         scheme_name=scheme_name,
     )

app/api/pagination.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class PaginatedList(list[T], Generic[T]):
+    """A page of items carrying the total count from its data source."""
+
+    def __init__(self, items: Iterable[T], *, total: int) -> None:
+        super().__init__(items)
+        self.total = total

app/api/v1/endpoints/audit.py

@@ -10,6 +10,7 @@ from app.auth.metadata_dependencies import (
     get_current_metadata_admin,
     get_current_metadata_user,
 )
+from app.api.pagination import PaginatedList
 from app.core.audit import AuditAction, log_audit_event
 from app.domain.schemas.audit import AuditLogResponse
 from app.infra.db.metadb.database import get_metadata_session
@@ -46,7 +47,7 @@ async def get_audit_logs(
     _current_user=Depends(get_current_metadata_admin),
     audit_repo: AuditRepository = Depends(get_audit_repository),
 ) -> list[AuditLogResponse]:
-    return await audit_repo.get_logs(
+    items = await audit_repo.get_logs(
         user_id=user_id,
         project_id=project_id,
         action=action,
@@ -56,6 +57,15 @@ async def get_audit_logs(
         skip=skip,
         limit=limit,
     )
+    total = await audit_repo.get_log_count(
+        user_id=user_id,
+        project_id=project_id,
+        action=action,
+        resource_type=resource_type,
+        start_time=start_time,
+        end_time=end_time,
+    )
+    return PaginatedList(items, total=total)
 
 
 @router.get(
@@ -119,7 +129,7 @@ async def get_my_audit_logs(
     current_user=Depends(get_current_metadata_user),
     audit_repo: AuditRepository = Depends(get_audit_repository),
 ) -> list[AuditLogResponse]:
-    return await audit_repo.get_logs(
+    items = await audit_repo.get_logs(
         user_id=current_user.id,
         action=action,
         start_time=start_time,
@@ -127,3 +137,10 @@ async def get_my_audit_logs(
         skip=skip,
         limit=limit,
     )
+    total = await audit_repo.get_log_count(
+        user_id=current_user.id,
+        action=action,
+        start_time=start_time,
+        end_time=end_time,
+    )
+    return PaginatedList(items, total=total)

app/api/v1/endpoints/simulation.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import Any, List, Literal, Optional
 from datetime import datetime, timedelta
 import json
 import threading
@@ -302,12 +302,20 @@ async def valve_isolation_endpoint(
     return result
 
 
-@router.post("/flushing-analyses", response_class=PlainTextResponse, summary="冲洗分析（高级）", description="高级版本的冲洗分析，支持同时开启多个阀门进行冲洗，指定排污节点，并设置固定的冲洗流量。返回纯文本格式的分析结果。")
+@router.post("/flushing-analyses", response_class=PlainTextResponse, summary="冲洗分析（高级）", description="高级版本的冲洗分析，支持按状态和设置值控制多个可选阀门，指定排污节点，并设置固定的冲洗流量。返回纯文本格式的分析结果。")
 async def fastapi_flushing_analysis(
     network: str = Query(..., description="管网名称（或数据库名称）"),
     start_time: str = Query(..., description="冲洗开始时间（ISO 8601格式）"),
-    valves: List[str] = Query(..., description="要开启的阀门ID列表"),
-    valves_k: List[float] = Query(..., description="对应各阀门的开度列表（0-1）"),
+    valves: List[str] | None = Query(None, description="参与控制的阀门ID列表（可选）"),
+    valves_k: List[float] | None = Query(
+        None, description="对应各阀门的开度列表（0-1，可选，与valves同时提供）"
+    ),
+    valve_statuses: List[Literal["OPEN", "CLOSED", "ACTIVE"]] | None = Query(
+        None, description="对应各阀门的开关状态列表（OPEN、CLOSED或ACTIVE，可选）"
+    ),
+    valve_settings: List[str] | None = Query(
+        None, description="对应各阀门的设置值列表（ACTIVE状态下必填）"
+    ),
     drainage_node_ID: str = Query(..., description="排污节点ID"),
     flush_flow: float = Query(0, description="冲洗流量（L/s），0表示自动计算"),
     duration: int | None = Query(None, description="模拟持续时间（秒），默认900秒"),
@@ -319,8 +327,10 @@ async def fastapi_flushing_analysis(
     
     - **network**: 管网名称（或数据库名称）
     - **start_time**: 冲洗开始时间
-    - **valves**: 要开启的阀门ID列表
-    - **valves_k**: 各阀门的开度列表（0-1，与valves对应）
+    - **valves**: 参与控制的阀门ID列表（可选）
+    - **valves_k**: 各阀门的开度列表（0-1，可选，与valves同时提供）
+    - **valve_statuses**: 各阀门的开关状态列表（OPEN、CLOSED或ACTIVE，可选）
+    - **valve_settings**: 各阀门的设置值列表（ACTIVE状态下必填）
     - **drainage_node_ID**: 排污节点ID
     - **flush_flow**: 冲洗流量（L/s）
     - **duration**: 模拟持续时间（秒，可选，默认900）
@@ -328,14 +338,75 @@ async def fastapi_flushing_analysis(
     
     支持多阀联合冲洗操作。
     """
+    valve_opening = None
+    valve_control = None
+    if valve_statuses is not None and valves_k is not None:
+        raise HTTPException(
+            status_code=422,
+            detail="valve_statuses 和 valves_k 不能同时提供",
+        )
+    if valve_settings is not None and valve_statuses is None:
+        raise HTTPException(
+            status_code=422,
+            detail="valve_settings 必须与 valve_statuses 同时提供",
+        )
+    if valves is None:
+        if (
+            valves_k is not None
+            or valve_statuses is not None
+            or valve_settings is not None
+        ):
+            raise HTTPException(
+                status_code=422,
+                detail="阀门控制参数必须与 valves 同时提供",
+            )
+    elif valve_statuses is not None:
+        if len(valves) != len(valve_statuses):
+            raise HTTPException(
+                status_code=422, detail="valves 和 valve_statuses 的数量必须一致"
+            )
+        if valve_settings is not None and len(valves) != len(valve_settings):
+            raise HTTPException(
+                status_code=422, detail="valves 和 valve_settings 的数量必须一致"
+            )
+
+        settings = valve_settings or [""] * len(valves)
+        valve_control = {}
+        for valve_id, raw_status, raw_setting in zip(
+            valves, valve_statuses, settings
+        ):
+            status = raw_status
+            setting = raw_setting.strip()
+            if status == "ACTIVE" and not setting:
+                raise HTTPException(
+                    status_code=422,
+                    detail=f"ACTIVE 状态的阀门 {valve_id} 必须提供设置值",
+                )
+
+            control: dict[str, str] = {"status": status}
+            if status == "ACTIVE":
+                control["setting"] = setting
+            valve_control[valve_id] = control
+    elif valves_k is not None:
+        if len(valves) != len(valves_k):
+            raise HTTPException(
+                status_code=422, detail="valves 和 valves_k 的数量必须一致"
+            )
         valve_opening = {
-        valve_id: float(valves_k[idx]) for idx, valve_id in enumerate(valves)
+            valve_id: float(valve_k)
+            for valve_id, valve_k in zip(valves, valves_k)
         }
+    else:
+        raise HTTPException(
+            status_code=422,
+            detail="提供 valves 时必须同时提供 valve_statuses 或 valves_k",
+        )
     result = flushing_analysis(
         name=network,
         modify_pattern_start_time=start_time,
         modify_total_duration=duration or 900,
         modify_valve_opening=valve_opening,
+        valve_control=valve_control,
         drainage_node_ID=drainage_node_ID,
         flushing_flow=flush_flow,
         scheme_name=scheme_name,

app/api/v1/rest_router.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, JsonValue, create_model
 from starlette.responses import Response
 
 from app.api.problem_details import ProblemDetails
+from app.api.pagination import PaginatedList
 from app.api.v1.router import api_router as handler_api_router
 from app.auth.metadata_dependencies import get_current_metadata_user
 from app.auth.project_dependencies import ProjectContext, get_project_context
@@ -41,8 +42,8 @@ _PUBLIC_PARAMETER_RENAMES = {
     "burst_ID": "burst_id",
     "drainage_node_ID": "drainage_node_id",
 }
-_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate"}
-_MODEL_USERNAME_FROM_AUTH: set[str] = set()
+_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate", "PressureSensorPlacement"}
+_MODEL_USERNAME_FROM_AUTH = {"PressureSensorPlacement"}
 
 
 def _clean_name(name: str) -> str:
@@ -230,9 +231,14 @@ def _with_pagination(endpoint):
         if not isinstance(result, list):
             return result
         if handler_handles_pagination:
+            if not isinstance(result, PaginatedList):
+                raise RuntimeError(
+                    f"Paginated handler {endpoint.__name__!r} must return "
+                    "PaginatedList with the real total"
+                )
             return Page(
                 items=result,
-                total=offset + len(result),
+                total=result.total,
                 limit=limit or len(result),
                 offset=offset,
             )

app/api/v1/router.py

@@ -53,6 +53,7 @@ from app.api.v1.endpoints.timeseries import (
 )
 from app.auth.permissions import (
     BURST_RUN,
+    ENVIRONMENT_MANAGE,
     OPTIMIZATION_RUN,
     RISK_RUN,
     SCADA_CLEAN,
@@ -88,6 +89,7 @@ simulation_access = Depends(
 
 webgis_view_access = Depends(require_permission(WEBGIS_VIEW))
 simulation_run_access = Depends(require_permission(SIMULATION_RUN))
+environment_manage_access = Depends(require_permission(ENVIRONMENT_MANAGE))
 burst_run_access = Depends(require_permission(BURST_RUN))
 risk_run_access = Depends(require_permission(RISK_RUN))
 optimization_run_access = Depends(require_permission(OPTIMIZATION_RUN))
@@ -169,7 +171,7 @@ api_router.include_router(
 api_router.include_router(
     cache.router,
     tags=["Cache"],
-    dependencies=[simulation_run_access],
+    dependencies=[environment_manage_access],
 )
 api_router.include_router(
     web_search.router,

app/core/audit.py

@@ -130,6 +130,7 @@ def sanitize_sensitive_data(data: dict) -> dict:
         "token",
         "api_key",
         "apikey",
+        "dsn",
         "credit_card",
         "ssn",
         "social_security",

app/infra/db/influxdb/info.py

@@ -1,6 +1,5 @@
-# influxdb数据库连接信息
-url = "http://127.0.0.1:8086"  # 替换为你的InfluxDB实例地址
-token = "kMPX2V5HsbzPpUT2B9HPBu1sTG1Emf-lPlT2UjxYnGAuocpXq_f_0lK4HHs-TbbKyjsZpICkMsyXG_V2D7P7yQ=="  # 替换为你的InfluxDB Token
-# _ENCODED_TOKEN = "eEdETTVSWnFSSkF1ekFHUy1vdFhVZEMyTkZkWTc1cUpBalJMcUFCNHA1V2NJSUFsSVVwT3BUOF95QTE2QU9IbUpXZXJ3UV8wOGd3Yjg0c3k0MmpuWlE9PQ=="
-# token = base64.b64decode(_ENCODED_TOKEN).decode("utf-8")
-org = "TJWATERORG"  # 替换为你的Organization名称
+from app.core.config import settings
+
+url = settings.INFLUXDB_URL
+token = settings.INFLUXDB_TOKEN
+org = settings.INFLUXDB_ORG

app/native/wndb/database.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping, Sequence
 from typing import Any
 from psycopg.rows import dict_row, Row
 from .connection import project_connection
@@ -82,27 +83,38 @@ class DbChangeSet:
         return DbChangeSet(redo_sql, undo_sql, redo_cs_s, undo_cs_s)
 
 
-def read(name: str, sql: str) -> Row:
+QueryParams = Sequence[Any] | Mapping[str, Any]
+
+
+def _execute(cur, sql: str, params: QueryParams | None = None):
+    return cur.execute(sql, params) if params is not None else cur.execute(sql)
+
+
+def read(name: str, sql: str, params: QueryParams | None = None) -> Row:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             row = cur.fetchone()
             if row == None:
                 raise Exception(sql)
             return row
 
 
-def read_all(name: str, sql: str) -> list[Row]:
+def read_all(
+    name: str, sql: str, params: QueryParams | None = None
+) -> list[Row]:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             return cur.fetchall()
 
 
-def try_read(name: str, sql: str) -> Row | None:
+def try_read(
+    name: str, sql: str, params: QueryParams | None = None
+) -> Row | None:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             return cur.fetchone()
 
 

app/native/wndb/s0_base.py

@@ -1,3 +1,4 @@
+from psycopg import sql
 from psycopg.rows import dict_row, Row
 from .connection import project_connection
 from .database import read
@@ -49,7 +50,12 @@ ELEMENT_TYPES : dict[str, int] = {
 def _get_from(name: str, id: str, base_type: str) -> Row | None:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(f"select * from {base_type} where id = '{id}'")
+            cur.execute(
+                sql.SQL("select * from {} where id = %s").format(
+                    sql.Identifier(base_type)
+                ),
+                (id,),
+            )
             return cur.fetchone()
 
 
@@ -243,11 +249,17 @@ def get_node_links(name: str, id: str) -> list[str]:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
             links: list[str] = []
-            for p in cur.execute(f"select id from pipes where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from pipes where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
-            for p in cur.execute(f"select id from pumps where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from pumps where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
-            for p in cur.execute(f"select id from valves where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from valves where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
             return links
 
@@ -255,16 +267,15 @@ def get_node_links(name: str, id: str) -> list[str]:
 def get_link_nodes(name: str, id: str) -> list[str]:
     row = {}
     if is_pipe(name, id):
-        row = read(name, f"select node1, node2 from pipes where id = '{id}'")
+        row = read(name, "select node1, node2 from pipes where id = %s", (id,))
     elif is_pump(name, id):
-        row = read(name, f"select node1, node2 from pumps where id = '{id}'")
+        row = read(name, "select node1, node2 from pumps where id = %s", (id,))
     elif is_valve(name, id):
-        row = read(name, f"select node1, node2 from valves where id = '{id}'")
+        row = read(name, "select node1, node2 from valves where id = %s", (id,))
     return [str(row['node1']), str(row['node2'])]
 
 def get_region_type(name: str, id: str)->str:
     if(is_region(name,id)):
-        type = read(name, f"select type from _region where id = '{id}'")
+        type = read(name, "select type from _region where id = %s", (id,))
         return type
 
-

app/native/wndb/s24_coordinates.py

@@ -23,7 +23,11 @@ def from_postgis_point(coord: str) -> dict[str, float]:
 
 
 def get_node_coord(name: str, node: str) -> dict[str, float]:
-    row = try_read(name, f"select st_astext(coord) as coord_geom from coordinates where node = '{node}'")
+    row = try_read(
+        name,
+        "select st_astext(coord) as coord_geom from coordinates where node = %s",
+        (node,),
+    )
     if row == None:
         write(name, sql_insert_coord(node, 0.0, 0.0))
         return {'x': 0.0, 'y': 0.0}
@@ -66,7 +70,9 @@ def get_links_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -
 
 
 def node_has_coord(name: str, node: str) -> bool:
-    return try_read(name, f"select node from coordinates where node = '{node}'") != None
+    return try_read(
+        name, "select node from coordinates where node = %s", (node,)
+    ) != None
 
 
 #--------------------------------------------------------------

app/native/wndb/s2_junctions.py

@@ -12,7 +12,7 @@ def get_junction_schema(name: str) -> dict[str, dict[str, Any]]:
 
 
 def get_junction(name: str, id: str) -> dict[str, Any]:
-    j = try_read(name, f"select * from junctions where id = '{id}'")
+    j = try_read(name, "select * from junctions where id = %s", (id,))
     if j == None:
         return {}
     xy = get_node_coord(name, id)

app/services/simulation.py

@@ -686,6 +686,28 @@ def get_history_pattern_info(project_name, pattern_name):
     return flow_list, factor_list
 
 
+def _apply_valve_control(
+    project_name: str, valve_control: dict[str, dict]
+) -> None:
+    """Apply explicit valve status, setting, and opening controls."""
+    for valve_name, control in valve_control.items():
+        valve_status = get_status(project_name, valve_name)
+        if "status" in control:
+            valve_status["status"] = control["status"]
+        if "setting" in control:
+            valve_status["setting"] = control["setting"]
+        if "k" in control:
+            valve_k = control["k"]
+            if valve_k == 0:
+                valve_status["status"] = "CLOSED"
+            else:
+                valve_status["setting"] = 0.1036 * pow(valve_k, -3.105)
+
+        cs = ChangeSet()
+        cs.append(valve_status)
+        set_status(project_name, cs)
+
+
 # 2025/01/11
 def run_simulation(
     name: str,
@@ -701,6 +723,7 @@ def run_simulation(
     modify_valve_opening: dict[str, float] = None,
     scheme_type: str = None,
     scheme_name: str = None,
+    valve_control: dict[str, dict] = None,
 ) -> None:
     """
     传入需要修改的参数，改变数据库中对应位置的值，然后计算，返回结果
@@ -715,6 +738,7 @@ def run_simulation(
     :param modify_fixed_pump_pattern: dict中包含多个水泵模式，str为工频水泵的id，list为修改后的pattern
     :param modify_variable_pump_pattern: dict中包含多个水泵模式，str为变频水泵的id，list为修改后的pattern
     :param modify_valve_opening: dict中包含多个阀门开启度，str为阀门的id，float为修改后的阀门开启度
+    :param valve_control: dict中可分别指定阀门的status、setting和k；存在时优先于modify_valve_opening
     :param scheme_type: 模拟方案类型
     :param scheme_name：模拟方案名称
     :return:
@@ -1200,8 +1224,11 @@ def run_simulation(
                 cs = ChangeSet()
                 cs.append(pump_pattern)
                 set_pattern(name_c, cs)
-    # 修改阀门（valve）的状态setting和status
-    if modify_valve_opening:
+    # 显式阀门控制沿用 run_simulation_ex 的处理顺序和覆盖规则。
+    if valve_control is not None:
+        _apply_valve_control(name_c, valve_control)
+    # 保留原开度参数逻辑，兼容现有方案调用。
+    elif modify_valve_opening:
         for valve_name in modify_valve_opening.keys():
             if not np.isnan(modify_valve_opening[valve_name]):
                 valve_status = get_status(name_c, valve_name)

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "b003a57c9b8c1a041b644363ef58afb8bfcede1fe076ce48a190d2a1ac915e3f"
+      "sha256": "df7ae927dcf5ae32c3c1ad9be3245b1b78b984ce1902dd91e6313770860e0d48"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -1859,7 +1859,7 @@
         "title": "PressureRegulationRest",
         "type": "object"
       },
-      "PressureSensorPlacement": {
+      "PressureSensorPlacementRest": {
         "properties": {
           "min_diameter": {
             "default": 0,
@@ -1867,11 +1867,6 @@
             "title": "Min Diameter",
             "type": "integer"
           },
-          "name": {
-            "description": "管网名称（或数据库名称）",
-            "title": "Name",
-            "type": "string"
-          },
           "scheme_name": {
             "description": "方案名称",
             "title": "Scheme Name",
@@ -1881,20 +1876,13 @@
             "description": "传感器数量",
             "title": "Sensor Number",
             "type": "integer"
-          },
-          "username": {
-            "description": "用户名",
-            "title": "Username",
-            "type": "string"
           }
         },
         "required": [
-          "name",
           "scheme_name",
-          "sensor_number",
-          "username"
+          "sensor_number"
         ],
-        "title": "PressureSensorPlacement",
+        "title": "PressureSensorPlacementRest",
         "type": "object"
       },
       "ProblemDetails": {
@@ -11038,7 +11026,7 @@
     },
     "/api/v1/flushing-analyses": {
       "post": {
-        "description": "高级版本的冲洗分析，支持同时开启多个阀门进行冲洗，指定排污节点，并设置固定的冲洗流量。返回纯文本格式的分析结果。",
+        "description": "高级版本的冲洗分析，支持按状态和设置值控制多个可选阀门，指定排污节点，并设置固定的冲洗流量。返回纯文本格式的分析结果。",
         "operationId": "post_flushing_analyses",
         "parameters": [
           {
@@ -11053,31 +11041,92 @@
             }
           },
           {
-            "description": "要开启的阀门ID列表",
+            "description": "参与控制的阀门ID列表（可选）",
             "in": "query",
             "name": "valves",
-            "required": true,
+            "required": false,
             "schema": {
-              "description": "要开启的阀门ID列表",
+              "anyOf": [
+                {
                   "items": {
                     "type": "string"
                   },
-              "title": "Valves",
                   "type": "array"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "参与控制的阀门ID列表（可选）",
+              "title": "Valves"
             }
           },
           {
-            "description": "对应各阀门的开度列表（0-1）",
+            "description": "对应各阀门的开度列表（0-1，可选，与valves同时提供）",
             "in": "query",
             "name": "valves_k",
-            "required": true,
+            "required": false,
             "schema": {
-              "description": "对应各阀门的开度列表（0-1）",
+              "anyOf": [
+                {
                   "items": {
                     "type": "number"
                   },
-              "title": "Valves K",
                   "type": "array"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "对应各阀门的开度列表（0-1，可选，与valves同时提供）",
+              "title": "Valves K"
+            }
+          },
+          {
+            "description": "对应各阀门的开关状态列表（OPEN、CLOSED或ACTIVE，可选）",
+            "in": "query",
+            "name": "valve_statuses",
+            "required": false,
+            "schema": {
+              "anyOf": [
+                {
+                  "items": {
+                    "enum": [
+                      "OPEN",
+                      "CLOSED",
+                      "ACTIVE"
+                    ],
+                    "type": "string"
+                  },
+                  "type": "array"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "对应各阀门的开关状态列表（OPEN、CLOSED或ACTIVE，可选）",
+              "title": "Valve Statuses"
+            }
+          },
+          {
+            "description": "对应各阀门的设置值列表（ACTIVE状态下必填）",
+            "in": "query",
+            "name": "valve_settings",
+            "required": false,
+            "schema": {
+              "anyOf": [
+                {
+                  "items": {
+                    "type": "string"
+                  },
+                  "type": "array"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "对应各阀门的设置值列表（ACTIVE状态下必填）",
+              "title": "Valve Settings"
             }
           },
           {
@@ -24833,7 +24882,7 @@
           "content": {
             "application/json": {
               "schema": {
-                "$ref": "#/components/schemas/PressureSensorPlacement",
+                "$ref": "#/components/schemas/PressureSensorPlacementRest",
                 "description": "传感器放置分析参数"
               }
             }
@@ -25073,7 +25122,7 @@
           "content": {
             "application/json": {
               "schema": {
-                "$ref": "#/components/schemas/PressureSensorPlacement",
+                "$ref": "#/components/schemas/PressureSensorPlacementRest",
                 "description": "传感器放置分析参数"
               }
             }

tests/api/test_audit_middleware.py

@@ -7,6 +7,21 @@ from fastapi.testclient import TestClient
 
 from app.infra.audit import middleware as audit_middleware
 from app.infra.audit.middleware import AuditMiddleware
+from app.core.audit import sanitize_sensitive_data
+
+
+def test_sanitize_sensitive_data_redacts_database_dsn() -> None:
+    raw_dsn = "postgresql://alice:supersecret@db.internal/project"
+
+    sanitized = sanitize_sensitive_data(
+        {"dsn": raw_dsn, "database": {"readonly_dsn": raw_dsn}}
+    )
+
+    assert sanitized == {
+        "dsn": "***REDACTED***",
+        "database": {"readonly_dsn": "***REDACTED***"},
+    }
+    assert raw_dsn not in str(sanitized)
 
 
 def test_post_streaming_response_survives_audit_body_capture(monkeypatch):

tests/api/test_openapi_contract.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 from pathlib import Path
+from unittest.mock import Mock
 from uuid import uuid4
 
 import pytest
@@ -11,8 +12,11 @@ from fastapi.testclient import TestClient
 
 from app.api.v1.endpoints import schemes as schemes_endpoint
 from app.api.v1.endpoints import simulation as simulation_endpoint
+from app.api.v1.endpoints import cache as cache_endpoint
+from app.api.pagination import PaginatedList
 from app.api.v1.rest_router import api_router, build_rest_router
 from app.api.v1.router import api_router as source_api_router
+from app.auth.metadata_dependencies import get_current_metadata_user
 from app.auth.project_dependencies import ProjectContext, get_project_context
 from scripts.check_openapi import current_contract_bytes, validate
 
@@ -130,6 +134,12 @@ def test_rest_contract_uses_header_project_context() -> None:
             assert "network" not in schema.get("properties", {})
             assert "network_name" not in schema.get("properties", {})
 
+    placement_schema = document["components"]["schemas"][
+        "PressureSensorPlacementRest"
+    ]
+    assert "name" not in placement_schema["properties"]
+    assert "username" not in placement_schema["properties"]
+
     assert "/api/v1/burst-analysis" not in document["paths"]
     assert "/api/v1/getpipeproperties/" not in document["paths"]
 
@@ -280,7 +290,7 @@ def test_rest_runtime_wraps_handler_paginated_list() -> None:
         limit: int = Query(2, ge=1, le=10),
     ) -> list[int]:
         records = [10, 20, 30, 40]
-        return records[skip : skip + limit]
+        return PaginatedList(records[skip : skip + limit], total=len(records))
 
     app = FastAPI(redirect_slashes=False)
     app.include_router(build_rest_router(source_router.routes), prefix="/api/v1")
@@ -293,12 +303,76 @@ def test_rest_runtime_wraps_handler_paginated_list() -> None:
     assert response.status_code == 200
     assert response.json() == {
         "items": [20, 30],
-        "total": 3,
+        "total": 4,
         "limit": 2,
         "offset": 1,
     }
 
 
+def test_sensor_placement_body_uses_authenticated_project_and_user(
+    monkeypatch,
+) -> None:
+    captured: dict[str, object] = {}
+
+    def fake_pressure_sensor_placement_kmeans(**kwargs):
+        captured.update(kwargs)
+
+    monkeypatch.setattr(
+        simulation_endpoint,
+        "pressure_sensor_placement_kmeans",
+        fake_pressure_sensor_placement_kmeans,
+    )
+    app = FastAPI(redirect_slashes=False)
+    app.include_router(api_router, prefix="/api/v1")
+    app.dependency_overrides[get_project_context] = lambda: ProjectContext(
+        project_id=uuid4(),
+        project_code="project_a",
+        user_id=uuid4(),
+        project_role="member",
+    )
+    app.dependency_overrides[get_current_metadata_user] = lambda: type(
+        "User", (), {"username": "alice"}
+    )()
+
+    response = TestClient(app, raise_server_exceptions=False).post(
+        "/api/v1/pressure-sensor-placement-kmeans",
+        json={
+            "scheme_name": "placement_01",
+            "sensor_number": 5,
+            "min_diameter": 100,
+        },
+    )
+
+    assert response.status_code == 200
+    assert captured == {
+        "name": "project_a",
+        "scheme_name": "placement_01",
+        "sensor_number": 5,
+        "min_diameter": 100,
+        "username": "alice",
+    }
+
+
+def test_cache_management_requires_environment_permission(monkeypatch) -> None:
+    flushdb = Mock(return_value=True)
+    monkeypatch.setattr(cache_endpoint.redis_client, "flushdb", flushdb)
+    app = FastAPI(redirect_slashes=False)
+    app.include_router(api_router, prefix="/api/v1")
+    app.dependency_overrides[get_project_context] = lambda: ProjectContext(
+        project_id=uuid4(),
+        project_code="project_a",
+        user_id=uuid4(),
+        project_role="member",
+    )
+
+    response = TestClient(app, raise_server_exceptions=False).delete(
+        "/api/v1/all-redis"
+    )
+
+    assert response.status_code == 403
+    flushdb.assert_not_called()
+
+
 def test_rest_runtime_json_encodes_untyped_datetime_response() -> None:
     source_router = APIRouter()
 

tests/api/test_simulation_endpoints.py

@@ -383,6 +383,7 @@ def test_flushing_endpoint_passes_required_scheme_name(monkeypatch):
         "modify_pattern_start_time": "2025-01-02T03:04:05+08:00",
         "modify_total_duration": 900,
         "modify_valve_opening": {"V1": 0.5},
+        "valve_control": None,
         "drainage_node_ID": "N1",
         "flushing_flow": 100.0,
         "scheme_name": "flush_case_01",
@@ -390,6 +391,145 @@ def test_flushing_endpoint_passes_required_scheme_name(monkeypatch):
     }
 
 
+def test_flushing_endpoint_allows_omitting_valves(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    captured = {}
+
+    def fake_flushing_analysis(**kwargs):
+        captured.update(kwargs)
+        return "ok"
+
+    monkeypatch.setattr(module, "flushing_analysis", fake_flushing_analysis)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params={
+            "network": "demo",
+            "start_time": "2025-01-02T03:04:05+08:00",
+            "drainage_node_ID": "N1",
+            "scheme_name": "flush_without_valves",
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.text == "ok"
+    assert captured["modify_valve_opening"] is None
+    assert captured["valve_control"] is None
+    assert captured["drainage_node_ID"] == "N1"
+
+
+def test_flushing_endpoint_passes_explicit_valve_control(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    captured = {}
+
+    def fake_flushing_analysis(**kwargs):
+        captured.update(kwargs)
+        return "ok"
+
+    monkeypatch.setattr(module, "flushing_analysis", fake_flushing_analysis)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params=[
+            ("network", "demo"),
+            ("start_time", "2025-01-02T03:04:05+08:00"),
+            ("valves", "V1"),
+            ("valves", "V2"),
+            ("valve_statuses", "ACTIVE"),
+            ("valve_statuses", "CLOSED"),
+            ("valve_settings", "2.5"),
+            ("valve_settings", ""),
+            ("drainage_node_ID", "N1"),
+            ("scheme_name", "flush_with_valve_control"),
+        ],
+    )
+
+    assert response.status_code == 200
+    assert captured["modify_valve_opening"] is None
+    assert captured["valve_control"] == {
+        "V1": {"status": "ACTIVE", "setting": "2.5"},
+        "V2": {"status": "CLOSED"},
+    }
+
+
+def test_flushing_endpoint_requires_setting_for_active_valve(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params={
+            "network": "demo",
+            "start_time": "2025-01-02T03:04:05+08:00",
+            "valves": "V1",
+            "valve_statuses": "ACTIVE",
+            "drainage_node_ID": "N1",
+            "scheme_name": "flush_without_active_setting",
+        },
+    )
+
+    assert response.status_code == 422
+
+
+def test_flushing_endpoint_rejects_mixed_valve_control_modes(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params={
+            "network": "demo",
+            "start_time": "2025-01-02T03:04:05+08:00",
+            "valves": "V1",
+            "valves_k": 0.5,
+            "valve_statuses": "ACTIVE",
+            "valve_settings": "2.5",
+            "drainage_node_ID": "N1",
+            "scheme_name": "flush_with_mixed_controls",
+        },
+    )
+
+    assert response.status_code == 422
+
+
+def test_flushing_endpoint_rejects_settings_without_statuses(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params={
+            "network": "demo",
+            "start_time": "2025-01-02T03:04:05+08:00",
+            "valves": "V1",
+            "valves_k": 0.5,
+            "valve_settings": "2.5",
+            "drainage_node_ID": "N1",
+            "scheme_name": "flush_with_orphan_settings",
+        },
+    )
+
+    assert response.status_code == 422
+
+
+def test_flushing_endpoint_requires_drainage_node(monkeypatch):
+    module = _load_simulation_module(monkeypatch)
+    client = _build_authenticated_client(module)
+
+    response = client.post(
+        "/api/v1/flushing-analyses",
+        params={
+            "network": "demo",
+            "start_time": "2025-01-02T03:04:05+08:00",
+            "scheme_name": "flush_without_drainage_node",
+        },
+    )
+
+    assert response.status_code == 422
+
+
 def test_contaminant_endpoint_passes_current_username(monkeypatch):
     module = _load_simulation_module(monkeypatch)
     captured = {}

tests/unit/test_scheme_simulation_timestep.py

@@ -1,3 +1,4 @@
+import inspect
 import json
 from datetime import timedelta
 
@@ -7,6 +8,70 @@ from app.infra.db.timescaledb.repositories.scheme import SchemeRepository
 from app.services.time_api import parse_utc_time
 
 
+def test_run_simulation_exposes_explicit_valve_control():
+    from app.services import simulation
+
+    parameters = inspect.signature(simulation.run_simulation).parameters
+
+    assert "valve_control" in parameters
+
+
+def test_apply_valve_control_matches_run_simulation_ex_semantics(monkeypatch):
+    from app.services import simulation
+
+    updates: dict[str, dict] = {}
+
+    monkeypatch.setattr(
+        simulation,
+        "get_status",
+        lambda project_name, valve_name: {
+            "link": valve_name,
+            "status": "OPEN",
+            "setting": 1.0,
+        },
+    )
+    monkeypatch.setattr(
+        simulation,
+        "set_status",
+        lambda project_name, changeset: updates.update(
+            {
+                changeset.operations[0]["link"]: changeset.operations[0].copy()
+            }
+        ),
+    )
+
+    simulation._apply_valve_control(
+        "demo",
+        {
+            "V-status": {"status": "ACTIVE"},
+            "V-setting": {"setting": 2.5},
+            "V-closed": {"status": "ACTIVE", "setting": 9.0, "k": 0},
+            "V-k": {"status": "ACTIVE", "setting": 9.0, "k": 0.5},
+        },
+    )
+
+    assert updates["V-status"] == {
+        "link": "V-status",
+        "status": "ACTIVE",
+        "setting": 1.0,
+    }
+    assert updates["V-setting"] == {
+        "link": "V-setting",
+        "status": "OPEN",
+        "setting": 2.5,
+    }
+    assert updates["V-closed"] == {
+        "link": "V-closed",
+        "status": "CLOSED",
+        "setting": 9.0,
+    }
+    assert updates["V-k"] == {
+        "link": "V-k",
+        "status": "ACTIVE",
+        "setting": 0.1036 * pow(0.5, -3.105),
+    }
+
+
 def _node_result(periods: int) -> list[dict]:
     return [
         {

tests/unit/test_wndb_query_safety.py

@@ -0,0 +1,21 @@
+from app.native.wndb import s2_junctions
+
+
+def test_get_junction_binds_untrusted_identifier(monkeypatch) -> None:
+    calls: list[tuple[str, str, tuple[str, ...]]] = []
+    malicious_id = "J-1'; DELETE FROM junctions; --"
+
+    def fake_try_read(name, statement, params):
+        calls.append((name, statement, params))
+        return None
+
+    monkeypatch.setattr(s2_junctions, "try_read", fake_try_read)
+
+    assert s2_junctions.get_junction("project_a", malicious_id) == {}
+    assert calls == [
+        (
+            "project_a",
+            "select * from junctions where id = %s",
+            (malicious_id,),
+        )
+    ]