feat(ci): 封装checkout逻辑为公共脚本scripts/ci_checkout.sh

.gitea/workflows/auto-merge.yml

@@ -16,46 +16,7 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              top_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == top_prefix[:-1]:
-                      continue
-                  if name.startswith(top_prefix):
-                      member.name = name[len(top_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          bash scripts/ci_checkout.sh
       - name: Auto merge develop PRs
         run: |
           bash scripts/auto_merge_prs.sh develop

.gitea/workflows/ci-cd.yml

@@ -15,6 +15,7 @@ on:
     branches:
       - main
       - develop
+  workflow_dispatch:
 
 permissions:
   contents: read
@@ -36,46 +37,7 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          bash scripts/ci_checkout.sh
       - name: Verify CI environment
         shell: sh
         run: |
@@ -84,6 +46,14 @@ jobs:
           python3 -m pip --version
           echo "CI environment is ready"
 
+      - name: Cache pip dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements*.txt') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
       - name: Install dependencies
         shell: sh
         run: |
@@ -281,45 +251,14 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
+          bash scripts/ci_checkout.sh
+      - name: Cache npm dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.npm
+          key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
+          restore-keys: |
+            ${{ runner.os }}-npm-
 
       - name: Install dependencies
         shell: sh
@@ -327,6 +266,7 @@ jobs:
           set -eu
           docker run --rm \
             -v "$PWD:/workspace" \
+            -v "$HOME/.npm:/root/.npm" \
             -w /workspace/apps/web \
             docker.m.daocloud.io/library/node:20 \
             sh -lc 'npm ci'
@@ -385,45 +325,7 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'INNERPY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        INNERPY
-
+        bash scripts/ci_checkout.sh
     - name: Build and push all images to Gitea Registry
       shell: sh
       env:
@@ -516,45 +418,15 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'PY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
+        bash scripts/ci_checkout.sh
+    - name: Cache npm dependencies
+      uses: actions/cache@v3
+      with:
+        path: ~/.npm
+        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
+        restore-keys: |
+          ${{ runner.os }}-npm-
+
     - name: Run Playwright E2E on staging
       shell: sh
       run: |
@@ -565,6 +437,7 @@ jobs:
           -e E2E_BROWSER_CHANNEL=chromium \
           -e PLAYWRIGHT_HEADLESS=1 \
           -v "$PWD:/workspace" \
+          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc "npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts"
@@ -583,45 +456,15 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'PY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
+        bash scripts/ci_checkout.sh
+    - name: Cache npm dependencies
+      uses: actions/cache@v3
+      with:
+        path: ~/.npm
+        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
+        restore-keys: |
+          ${{ runner.os }}-npm-
+
     - name: Run API integration tests on staging
       shell: sh
       run: |
@@ -630,6 +473,7 @@ jobs:
           -e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
           -e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
           -v "$PWD:/workspace" \
+          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts'
@@ -650,46 +494,7 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'PY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
-
+        bash scripts/ci_checkout.sh
     - name: Build and push all images (api + worker + web, with buildx cache)
       shell: sh
       env:
@@ -793,46 +598,14 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        python3 - <<'PY'
-        import io, os, tarfile, time, urllib.request, urllib.error
-        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-        # Retry up to 5 times with backoff for transient 5xx errors
-        last_err = None
-        for attempt in range(5):
-            try:
-                with urllib.request.urlopen(request, timeout=120) as response:
-                    archive = response.read()
-                break
-            except urllib.error.HTTPError as e:
-                last_err = e
-                if e.code >= 500 and attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-            except Exception as e:
-                last_err = e
-                if attempt < 4:
-                    wait = 2 ** attempt
-                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                    time.sleep(wait)
-                    continue
-                raise
-        else:
-            raise last_err
-        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-            for member in tar.getmembers():
-                name = member.name
-                if name == root_prefix[:-1]:
-                    continue
-                if name.startswith(root_prefix):
-                    member.name = name[len(root_prefix):]
-                if member.name:
-                    tar.extract(member, '.')
-        PY
+        bash scripts/ci_checkout.sh
+    - name: Cache npm dependencies
+      uses: actions/cache@v3
+      with:
+        path: ~/.npm
+        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
+        restore-keys: |
+          ${{ runner.os }}-npm-
 
     - name: Run production browser E2E
       shell: sh
@@ -844,6 +617,7 @@ jobs:
           -e E2E_BROWSER_CHANNEL=chromium \
           -e PLAYWRIGHT_HEADLESS=1 \
           -v "$PWD:/workspace" \
+          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts'

.gitea/workflows/daily-check.yml

@@ -24,46 +24,7 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          bash scripts/ci_checkout.sh
       - name: Production health check & smoke test
         id: smoke
         shell: sh
@@ -121,46 +82,7 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          bash scripts/ci_checkout.sh
       - name: Run API smoke test on staging
         id: smoke
         shell: sh
@@ -258,45 +180,14 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
+          bash scripts/ci_checkout.sh
+      - name: Cache npm dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.npm
+          key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
+          restore-keys: |
+            ${{ runner.os }}-npm-
 
       - name: Run Playwright E2E on staging
         id: e2e
@@ -310,6 +201,7 @@ jobs:
             -e E2E_BROWSER_CHANNEL=chromium \
             -e PLAYWRIGHT_HEADLESS=1 \
             -v "$PWD:/workspace" \
+            -v "$HOME/.npm:/root/.npm" \
             -w /workspace/apps/web \
             git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
             sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts' 2>&1 | tee /tmp/staging-e2e.log

apps/worker/worker_app/tasks/generation.py

@@ -14,7 +14,6 @@ from urllib.parse import urlparse
 
 import oss2
 from worker_app.celery_app import celery_app
-from worker_app.db import SessionLocal
 
 OUTPUT_WIDTH = 1280
 OUTPUT_HEIGHT = 720
@@ -29,58 +28,6 @@ PUBLIC_API_BASE_URL = os.getenv("PUBLIC_API_BASE_URL", "https://api.xiaoxiajianj
 logger = logging.getLogger(__name__)
 
 
-# ── 状态更新辅助函数 ──────────────────────────────────────────────────────────
-
-
-def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
-    """更新 GenerationTask 状态（独立 session，异常不向外抛出）。
-
-    Args:
-        task_id: 任务 ID
-        status_action: 状态动作名，如 "mark_processing" / "mark_completed" / "mark_failed"
-        **kwargs: 传递给对应方法的参数
-
-    Returns:
-        True 表示更新成功，False 表示更新失败
-    """
-    try:
-        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-            SQLAlchemyGenerationTaskRepository,
-        )
-
-        session = SessionLocal()
-        try:
-            repo = SQLAlchemyGenerationTaskRepository(session)
-            task = repo.get(task_id)
-            if task is None:
-                logger.warning("更新任务状态失败：任务不存在 task_id=%s", task_id)
-                return False
-
-            action = getattr(task, status_action, None)
-            if action is None:
-                logger.warning("未知的状态动作: %s", status_action)
-                return False
-
-            action(**kwargs)
-            repo.update(task)
-            logger.info("GenerationTask 状态更新成功: task_id=%s action=%s", task_id, status_action)
-            return True
-        finally:
-            session.close()
-    except Exception as e:
-        logger.error(
-            "更新 GenerationTask 状态异常: task_id=%s action=%s error=%s",
-            task_id,
-            status_action,
-            e,
-            exc_info=True,
-        )
-        return False
-
-
-# ── FFmpeg / OSS helpers ─────────────────────────────────────────────────────
-
-
 def _run_ffmpeg(command: list[str]) -> None:
     """执行 FFmpeg 命令"""
     subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)  # nosec B603
@@ -208,6 +155,8 @@ def _download_library_assets(
     """
     # 导入模型和会话
     try:
+        from worker_app.db import SessionLocal
+
         from packages.adapters.sqlalchemy_impl.models import AssetModel
 
         session = SessionLocal()
@@ -282,9 +231,6 @@ def _process_with_editing_mode(
     )
 
 
-# ── Celery Task ──────────────────────────────────────────────────────────────
-
-
 @celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
 def generate_video(self, task_id: str) -> dict:
     """
@@ -296,21 +242,19 @@ def generate_video(self, task_id: str) -> dict:
     Returns:
         生成结果字典
     """
-    from packages.domain import EditingMode
+    from worker_app.db import SessionLocal
 
-    logger.info("开始生成视频任务: task_id=%s", task_id)
+    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+        SQLAlchemyGenerationTaskRepository,
+    )
+    from packages.domain import EditingMode, GeneratedVideo, GenerationTaskStatus
 
     # 从数据库加载任务信息
     session = SessionLocal()
     try:
-        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
-            SQLAlchemyGenerationTaskRepository,
-        )
-
         task_repo = SQLAlchemyGenerationTaskRepository(session)
         gen_task = task_repo.get(task_id)
         if gen_task is None:
-            logger.error("生成任务不存在: task_id=%s", task_id)
             return {"status": "failed", "error": f"generation task {task_id} not found"}
         project_id = gen_task.project_id
         asset_library_id = gen_task.asset_library_id
@@ -321,9 +265,6 @@ def generate_video(self, task_id: str) -> dict:
     finally:
         session.close()
 
-    # 标记任务为 running
-    _update_task_status(task_id, "mark_processing")
-
     try:
         editing_mode = EditingMode(mode)
     except ValueError:
@@ -374,7 +315,7 @@ def generate_video(self, task_id: str) -> dict:
                 file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}"
 
             # 创建 GeneratedVideo 记录 + 查重
-            video_count = _create_video_record_and_dedup(
+            _create_video_record_and_dedup(
                 task_id=task_id,
                 project_id=project_id,
                 batch_id=batch_id,
@@ -385,11 +326,6 @@ def generate_video(self, task_id: str) -> dict:
                 mode=editing_mode.value,
             )
 
-            # 标记任务为 completed
-            _update_task_status(task_id, "mark_completed", result_count=video_count or 1)
-
-            logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size)
-
             return {
                 "status": "completed",
                 "task_id": task_id,
@@ -401,9 +337,7 @@ def generate_video(self, task_id: str) -> dict:
                 "mode": editing_mode.value,
             }
     except Exception as error:
-        logger.error(f"Video generation failed: {error}", exc_info=True)
-        # 标记任务为 failed
-        _update_task_status(task_id, "mark_failed", error_message=str(error))
+        logger.error(f"Video generation failed: {error}")
         return {
             "status": "failed",
             "task_id": task_id,
@@ -421,15 +355,12 @@ def _create_video_record_and_dedup(
     duration: float,
     video_path: str,
     mode: str,
-) -> int:
-    """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。
-
-    Returns:
-        创建的视频记录数量（1 表示成功，0 表示失败）
-    """
+) -> None:
+    """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。"""
     from uuid import uuid4
 
     from video_processing.dedup import VideoDeduplicator
+    from worker_app.db import SessionLocal
 
     from packages.adapters.sqlalchemy_impl.generated_video_repository import (
         SQLAlchemyGeneratedVideoRepository,
@@ -464,7 +395,7 @@ def _create_video_record_and_dedup(
         except Exception as fp_err:
             logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}")
             session.commit()
-            return 1
+            return
 
         generated_video.video_fingerprint = fingerprint.to_dict()
 
@@ -489,10 +420,8 @@ def _create_video_record_and_dedup(
         video_repo.update(generated_video)
         session.commit()
         logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})")
-        return 1
     except Exception as e:
         logger.error(f"Failed to create video record / dedup for task {task_id}: {e}")
         session.rollback()
-        return 0
     finally:
         session.close()

packages/domain/generation_task.py

@@ -1,11 +1,3 @@
-"""GenerationTask 领域模型 — 视频生成任务.
-
-状态机：
-  pending → running → completed
-              ↘ failed → pending (重试)
-       ↘ cancelled
-"""
-
 from __future__ import annotations
 
 import sys
@@ -25,43 +17,11 @@ from uuid import uuid4
 
 
 class GenerationTaskStatus(StrEnum):
-    """生成任务状态枚举。"""
-
     PENDING = "pending"
-    """待处理（任务已创建，等待执行）"""
-
     RUNNING = "running"
-    """运行中（正在生成视频）"""
-
     COMPLETED = "completed"
-    """已完成（视频生成成功）"""
-
     FAILED = "failed"
-    """失败（生成失败）"""
-
     CANCELLED = "cancelled"
-    """已取消（用户取消或系统取消）"""
-
-
-# 终态集合
-TERMINAL_STATUSES = frozenset(
-    {GenerationTaskStatus.COMPLETED, GenerationTaskStatus.FAILED, GenerationTaskStatus.CANCELLED}
-)
-
-# 合法状态转换
-_VALID_TRANSITIONS: dict[GenerationTaskStatus, set[GenerationTaskStatus]] = {
-    GenerationTaskStatus.PENDING: {
-        GenerationTaskStatus.RUNNING,
-        GenerationTaskStatus.FAILED,
-        GenerationTaskStatus.CANCELLED,
-    },
-    GenerationTaskStatus.RUNNING: {
-        GenerationTaskStatus.COMPLETED,
-        GenerationTaskStatus.FAILED,
-        GenerationTaskStatus.CANCELLED,
-    },
-    GenerationTaskStatus.FAILED: {GenerationTaskStatus.PENDING},  # 重试回到 pending
-}
 
 
 @dataclass(slots=True)
@@ -123,123 +83,3 @@ class GenerationTask:
             asset_select_mode=asset_select_mode,
             batch_id=batch_id,
         )
-
-    # ── 状态查询 ────────────────────────────────────────────────────────────
-
-    @property
-    def is_terminal(self) -> bool:
-        """是否处于终态（completed / failed / cancelled）。"""
-        return self.status in TERMINAL_STATUSES
-
-    @property
-    def is_completed(self) -> bool:
-        """是否已完成。"""
-        return self.status == GenerationTaskStatus.COMPLETED
-
-    @property
-    def is_failed(self) -> bool:
-        """是否失败。"""
-        return self.status == GenerationTaskStatus.FAILED
-
-    @property
-    def is_running(self) -> bool:
-        """是否运行中。"""
-        return self.status == GenerationTaskStatus.RUNNING
-
-    # ── 状态转换 ────────────────────────────────────────────────────────────
-
-    def transition_to(self, new_status: GenerationTaskStatus | str) -> None:
-        """执行状态转换。
-
-        Args:
-            new_status: 目标状态
-
-        Raises:
-            ValueError: 非法状态转换
-        """
-        if isinstance(new_status, str):
-            try:
-                new_status = GenerationTaskStatus(new_status)
-            except ValueError:
-                raise ValueError(f"无效状态: {new_status}")
-
-        allowed = _VALID_TRANSITIONS.get(self.status, set())
-        if new_status not in allowed:
-            raise ValueError(
-                f"非法状态转换: {self.status.value} → {new_status.value}，"
-                f"允许: {{{', '.join(sorted(s.value for s in allowed))}}}"
-            )
-
-        self.status = new_status
-
-    def mark_processing(self) -> None:
-        """标记为处理中（pending → running）。
-
-        设置 started_at，清除 error_message。
-
-        Raises:
-            ValueError: 当前状态不允许转换到 running
-        """
-        self.transition_to(GenerationTaskStatus.RUNNING)
-        self.started_at = datetime.now(timezone.utc)
-        self.error_message = ""
-
-    def mark_completed(self, result_count: int = 1) -> None:
-        """标记为已完成（running → completed）。
-
-        设置 completed_at、progress=100.0、result_count，清除 error_message。
-
-        Args:
-            result_count: 生成的视频数量，默认为 1
-
-        Raises:
-            ValueError: 当前状态不允许转换到 completed
-        """
-        self.transition_to(GenerationTaskStatus.COMPLETED)
-        self.completed_at = datetime.now(timezone.utc)
-        self.progress = 100.0
-        self.result_count = result_count
-        self.error_message = ""
-
-    def mark_failed(self, error_message: str) -> None:
-        """标记为失败（pending / running → failed）。
-
-        设置 error_message、completed_at。
-
-        Args:
-            error_message: 错误信息
-
-        Raises:
-            ValueError: 当前状态不允许转换到 failed
-        """
-        self.transition_to(GenerationTaskStatus.FAILED)
-        self.error_message = error_message
-        self.completed_at = datetime.now(timezone.utc)
-
-    def mark_cancelled(self) -> None:
-        """标记为已取消（pending / running → cancelled）。
-
-        设置 completed_at。
-
-        Raises:
-            ValueError: 当前状态不允许转换到 cancelled
-        """
-        self.transition_to(GenerationTaskStatus.CANCELLED)
-        self.completed_at = datetime.now(timezone.utc)
-
-    def mark_pending_from_failed(self) -> None:
-        """从失败状态重置为待处理（用于重试）。
-
-        清除 error_message、started_at、completed_at、progress。
-
-        Raises:
-            ValueError: 当前状态不是 failed
-        """
-        if self.status != GenerationTaskStatus.FAILED:
-            raise ValueError(f"只有 failed 状态的任务可以重置为 pending，当前状态: {self.status.value}")
-        self.transition_to(GenerationTaskStatus.PENDING)
-        self.error_message = ""
-        self.started_at = None
-        self.completed_at = None
-        self.progress = 0.0
-        self.result_count = 0

scripts/ci_checkout.sh

@@ -0,0 +1,97 @@
+#!/bin/sh
+# CI Checkout script - 从 Gitea API 下载源码 tar 包并解压
+# 用法: ci_checkout.sh [repo_api_base] [ref] [target_dir] [token]
+#   repo_api_base: 仓库 API 基础 URL，如 https://git.xiaoxiajianji.com/api/v1/repos/xiaoxia/xiaoxia-saas
+#   ref:           commit SHA 或分支名
+#   target_dir:    目标目录（默认当前目录）
+#   token:         API token
+# 所有参数均可省略，将从 Gitea Actions 环境变量中读取
+
+set -eu
+
+# ── 参数解析 ──────────────────────────────────────────────────
+REPO_API_BASE="${1:-}"
+REF="${2:-}"
+TARGET_DIR="${3:-.}"
+TOKEN="${4:-}"
+
+# 从环境变量补全默认值（兼容 Gitea Actions）
+if [ -z "$REPO_API_BASE" ]; then
+  REPO_API_BASE="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}"
+fi
+if [ -z "$REF" ]; then
+  REF="${GITHUB_SHA}"
+fi
+if [ -z "$TOKEN" ]; then
+  TOKEN="${GITHUB_TOKEN:-}"
+fi
+
+if [ -z "$REPO_API_BASE" ] || [ -z "$REF" ]; then
+  echo "ERROR: repo API base and ref are required" >&2
+  echo "Usage: $0 [repo_api_base] [ref] [target_dir] [token]" >&2
+  exit 1
+fi
+
+# ── 下载并解压 ────────────────────────────────────────────────
+ARCHIVE_URL="${REPO_API_BASE}/archive/${REF}.tar.gz"
+
+echo "Checkout: ${ARCHIVE_URL}"
+echo "Target dir: ${TARGET_DIR}"
+
+mkdir -p "${TARGET_DIR}"
+
+# 通过环境变量传递给 Python
+_CHECKOUT_URL="${ARCHIVE_URL}" \
+_CHECKOUT_TOKEN="${TOKEN}" \
+_CHECKOUT_TARGET_DIR="${TARGET_DIR}" \
+python3 - <<'PY'
+import io, os, tarfile, time, urllib.request, urllib.error
+
+url = os.environ['_CHECKOUT_URL']
+token = os.environ.get('_CHECKOUT_TOKEN', '')
+target_dir = os.environ['_CHECKOUT_TARGET_DIR']
+
+headers = {}
+if token:
+    headers["Authorization"] = f"token {token}"
+
+request = urllib.request.Request(url, headers=headers)
+
+last_err = None
+for attempt in range(5):
+    try:
+        with urllib.request.urlopen(request, timeout=120) as response:
+            archive = response.read()
+        break
+    except urllib.error.HTTPError as e:
+        last_err = e
+        if e.code >= 500 and attempt < 4:
+            wait = 2 ** attempt
+            print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+            time.sleep(wait)
+            continue
+        raise
+    except Exception as e:
+        last_err = e
+        if attempt < 4:
+            wait = 2 ** attempt
+            print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+            time.sleep(wait)
+            continue
+        raise
+else:
+    raise last_err
+
+with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+    root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+    for member in tar.getmembers():
+        name = member.name
+        if name == root_prefix[:-1]:
+            continue
+        if name.startswith(root_prefix):
+            member.name = name[len(root_prefix):]
+        if member.name:
+            tar.extract(member, target_dir)
+
+print("Checkout complete.")
+PY

tests/unit/test_generation_task_status.py

@@ -1,455 +0,0 @@
-"""GenerationTask 领域模型状态机单元测试.
-
-覆盖：
-- 初始状态为 pending
-- mark_processing: pending → running
-- mark_completed: running → completed
-- mark_failed: pending/running → failed
-- mark_cancelled: pending/running → cancelled
-- mark_pending_from_failed: failed → pending（重试）
-- 非法状态转换抛出 ValueError
-- is_terminal / is_completed / is_failed / is_running 属性
-- 状态转换时的时间戳设置
-"""
-
-from __future__ import annotations
-
-import pytest
-
-from packages.domain.generation_task import (
-    TERMINAL_STATUSES,
-    GenerationTask,
-    GenerationTaskStatus,
-)
-
-
-def _make_task(**overrides) -> GenerationTask:
-    """创建一个测试用的 GenerationTask。"""
-    defaults = dict(
-        id="task-test-001",
-        project_id="proj-1",
-        asset_library_id="lib-1",
-    )
-    defaults.update(overrides)
-    return GenerationTask(**defaults)
-
-
-# ── 初始状态 ──────────────────────────────────────────────────────────────────
-
-
-class TestInitialState:
-    """测试初始状态。"""
-
-    def test_default_status_is_pending(self) -> None:
-        """新创建的任务默认状态为 pending。"""
-        task = _make_task()
-        assert task.status == GenerationTaskStatus.PENDING
-        assert task.progress == 0.0
-        assert task.result_count == 0
-        assert task.error_message == ""
-        assert task.started_at is None
-        assert task.completed_at is None
-
-    def test_create_factory_returns_pending(self) -> None:
-        """GenerationTask.create() 返回的任务状态为 pending。"""
-        task = GenerationTask.create(
-            project_id="proj-1",
-            asset_library_id="lib-1",
-            created_by_user_id="user-1",
-        )
-        assert task.status == GenerationTaskStatus.PENDING
-        assert task.progress == 0.0
-        assert task.result_count == 0
-
-    def test_is_not_terminal_initially(self) -> None:
-        """初始状态不是终态。"""
-        task = _make_task()
-        assert not task.is_terminal
-        assert not task.is_completed
-        assert not task.is_failed
-        assert not task.is_running
-
-    def test_terminal_statuses_constant(self) -> None:
-        """终态集合包含 completed / failed / cancelled。"""
-        assert GenerationTaskStatus.COMPLETED in TERMINAL_STATUSES
-        assert GenerationTaskStatus.FAILED in TERMINAL_STATUSES
-        assert GenerationTaskStatus.CANCELLED in TERMINAL_STATUSES
-        assert GenerationTaskStatus.PENDING not in TERMINAL_STATUSES
-        assert GenerationTaskStatus.RUNNING not in TERMINAL_STATUSES
-
-
-# ── mark_processing ──────────────────────────────────────────────────────────
-
-
-class TestMarkProcessing:
-    """测试 pending → running 转换。"""
-
-    def test_pending_to_running_success(self) -> None:
-        """pending 状态的任务可以标记为 running。"""
-        task = _make_task()
-        task.mark_processing()
-        assert task.status == GenerationTaskStatus.RUNNING
-        assert task.is_running
-        assert task.started_at is not None
-        assert task.error_message == ""
-
-    def test_started_at_is_set(self) -> None:
-        """mark_processing 设置 started_at 时间戳。"""
-        task = _make_task()
-        assert task.started_at is None
-        task.mark_processing()
-        assert task.started_at is not None
-
-    def test_error_message_cleared(self) -> None:
-        """mark_processing 清除 error_message（如果有的话）。"""
-        task = _make_task()
-        # 注意：pending 状态通常没有 error_message，这里验证确保被清除
-        task.error_message = "some old error"
-        # 直接设置状态绕过校验（模拟异常场景）
-        task.status = GenerationTaskStatus.PENDING
-        task.mark_processing()
-        assert task.error_message == ""
-
-    def test_running_to_running_raises(self) -> None:
-        """running 状态不能再次 mark_processing。"""
-        task = _make_task()
-        task.mark_processing()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_processing()
-
-    def test_completed_to_running_raises(self) -> None:
-        """completed 状态不能回到 running。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_processing()
-
-    def test_failed_to_running_raises(self) -> None:
-        """failed 状态不能直接到 running（应先重置为 pending）。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_failed("some error")
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_processing()
-
-
-# ── mark_completed ───────────────────────────────────────────────────────────
-
-
-class TestMarkCompleted:
-    """测试 running → completed 转换。"""
-
-    def test_running_to_completed_success(self) -> None:
-        """running 状态的任务可以标记为 completed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        assert task.status == GenerationTaskStatus.COMPLETED
-        assert task.is_completed
-        assert task.is_terminal
-        assert task.completed_at is not None
-
-    def test_progress_set_to_100(self) -> None:
-        """mark_completed 设置 progress 为 100.0。"""
-        task = _make_task()
-        task.mark_processing()
-        task.progress = 50.0  # 模拟中间进度
-        task.mark_completed()
-        assert task.progress == 100.0
-
-    def test_default_result_count_is_1(self) -> None:
-        """默认 result_count 为 1。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        assert task.result_count == 1
-
-    def test_custom_result_count(self) -> None:
-        """可以指定 result_count。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed(result_count=5)
-        assert task.result_count == 5
-
-    def test_error_message_cleared(self) -> None:
-        """mark_completed 清除 error_message。"""
-        task = _make_task()
-        task.mark_processing()
-        task.error_message = "temporary error"
-        task.mark_completed()
-        assert task.error_message == ""
-
-    def test_completed_at_is_set(self) -> None:
-        """mark_completed 设置 completed_at。"""
-        task = _make_task()
-        task.mark_processing()
-        assert task.completed_at is None
-        task.mark_completed()
-        assert task.completed_at is not None
-
-    def test_pending_to_completed_raises(self) -> None:
-        """pending 状态不能直接到 completed。"""
-        task = _make_task()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_completed()
-
-    def test_completed_to_completed_raises(self) -> None:
-        """completed 状态不能再次 mark_completed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_completed()
-
-    def test_failed_to_completed_raises(self) -> None:
-        """failed 状态不能直接到 completed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_failed("error")
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_completed()
-
-
-# ── mark_failed ──────────────────────────────────────────────────────────────
-
-
-class TestMarkFailed:
-    """测试 pending/running → failed 转换。"""
-
-    def test_pending_to_failed_success(self) -> None:
-        """pending 状态可以直接标记为 failed。"""
-        task = _make_task()
-        task.mark_failed("资源不足")
-        assert task.status == GenerationTaskStatus.FAILED
-        assert task.is_failed
-        assert task.is_terminal
-        assert task.error_message == "资源不足"
-        assert task.completed_at is not None
-
-    def test_running_to_failed_success(self) -> None:
-        """running 状态可以标记为 failed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_failed("生成失败：FFmpeg 错误")
-        assert task.status == GenerationTaskStatus.FAILED
-        assert task.is_failed
-        assert task.is_terminal
-        assert task.error_message == "生成失败：FFmpeg 错误"
-        assert task.completed_at is not None
-
-    def test_completed_to_failed_raises(self) -> None:
-        """completed 状态不能标记为 failed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_failed("late error")
-
-    def test_failed_to_failed_raises(self) -> None:
-        """failed 状态不能再次 mark_failed。"""
-        task = _make_task()
-        task.mark_failed("first error")
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_failed("second error")
-
-    def test_error_message_preserved(self) -> None:
-        """错误信息被正确保存。"""
-        task = _make_task()
-        error_msg = "FFmpeg returned non-zero exit status 1"
-        task.mark_failed(error_msg)
-        assert task.error_message == error_msg
-
-
-# ── mark_cancelled ───────────────────────────────────────────────────────────
-
-
-class TestMarkCancelled:
-    """测试 pending/running → cancelled 转换。"""
-
-    def test_pending_to_cancelled_success(self) -> None:
-        """pending 状态可以取消。"""
-        task = _make_task()
-        task.mark_cancelled()
-        assert task.status == GenerationTaskStatus.CANCELLED
-        assert task.is_terminal
-        assert task.completed_at is not None
-
-    def test_running_to_cancelled_success(self) -> None:
-        """running 状态可以取消。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_cancelled()
-        assert task.status == GenerationTaskStatus.CANCELLED
-        assert task.is_terminal
-
-    def test_completed_to_cancelled_raises(self) -> None:
-        """completed 状态不能取消。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_cancelled()
-
-    def test_failed_to_cancelled_raises(self) -> None:
-        """failed 状态不能取消。"""
-        task = _make_task()
-        task.mark_failed("some error")
-        with pytest.raises(ValueError, match="非法状态转换"):
-            task.mark_cancelled()
-
-
-# ── mark_pending_from_failed (重试) ─────────────────────────────────────────
-
-
-class TestMarkPendingFromFailed:
-    """测试 failed → pending（重试）转换。"""
-
-    def test_failed_to_pending_success(self) -> None:
-        """failed 状态可以重置为 pending（用于重试）。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_failed("临时错误")
-        task.mark_pending_from_failed()
-        assert task.status == GenerationTaskStatus.PENDING
-        assert not task.is_terminal
-        assert task.error_message == ""
-        assert task.started_at is None
-        assert task.completed_at is None
-        assert task.progress == 0.0
-        assert task.result_count == 0
-
-    def test_pending_to_pending_raises(self) -> None:
-        """pending 状态不能调用 mark_pending_from_failed。"""
-        task = _make_task()
-        with pytest.raises(ValueError, match="只有 failed 状态"):
-            task.mark_pending_from_failed()
-
-    def test_running_to_pending_raises(self) -> None:
-        """running 状态不能调用 mark_pending_from_failed。"""
-        task = _make_task()
-        task.mark_processing()
-        with pytest.raises(ValueError, match="只有 failed 状态"):
-            task.mark_pending_from_failed()
-
-    def test_completed_to_pending_raises(self) -> None:
-        """completed 状态不能调用 mark_pending_from_failed。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError, match="只有 failed 状态"):
-            task.mark_pending_from_failed()
-
-
-# ── transition_to 通用方法 ───────────────────────────────────────────────────
-
-
-class TestTransitionTo:
-    """测试通用的 transition_to 方法。"""
-
-    def test_string_status_conversion(self) -> None:
-        """可以传入字符串形式的状态。"""
-        task = _make_task()
-        task.transition_to("running")
-        assert task.status == GenerationTaskStatus.RUNNING
-
-    def test_invalid_string_raises(self) -> None:
-        """无效的状态字符串抛出 ValueError。"""
-        task = _make_task()
-        with pytest.raises(ValueError, match="无效状态"):
-            task.transition_to("invalid_status")
-
-    def test_enum_status(self) -> None:
-        """可以传入枚举形式的状态。"""
-        task = _make_task()
-        task.transition_to(GenerationTaskStatus.RUNNING)
-        assert task.status == GenerationTaskStatus.RUNNING
-
-    def test_error_message_includes_allowed_statuses(self) -> None:
-        """错误信息包含允许的状态列表。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_completed()
-        with pytest.raises(ValueError) as exc_info:
-            task.transition_to(GenerationTaskStatus.RUNNING)
-        assert "completed" in str(exc_info.value)
-        assert "running" in str(exc_info.value)
-
-
-# ── 完整流转路径 ─────────────────────────────────────────────────────────────
-
-
-class TestFullFlow:
-    """测试完整的状态流转路径。"""
-
-    def test_happy_path(self) -> None:
-        """正常路径：pending → running → completed。"""
-        task = _make_task()
-        assert task.status == GenerationTaskStatus.PENDING
-        assert not task.is_terminal
-
-        task.mark_processing()
-        assert task.status == GenerationTaskStatus.RUNNING
-        assert task.started_at is not None
-        assert not task.is_terminal
-
-        task.mark_completed(result_count=3)
-        assert task.status == GenerationTaskStatus.COMPLETED
-        assert task.is_completed
-        assert task.is_terminal
-        assert task.completed_at is not None
-        assert task.result_count == 3
-        assert task.progress == 100.0
-
-    def test_failure_path_from_running(self) -> None:
-        """失败路径：pending → running → failed。"""
-        task = _make_task()
-        task.mark_processing()
-        assert task.is_running
-
-        task.mark_failed("网络超时")
-        assert task.is_failed
-        assert task.is_terminal
-        assert task.error_message == "网络超时"
-        assert task.completed_at is not None
-
-    def test_failure_path_from_pending(self) -> None:
-        """失败路径：pending → failed（启动前校验失败等）。"""
-        task = _make_task()
-        task.mark_failed("参数校验失败")
-        assert task.is_failed
-        assert task.is_terminal
-
-    def test_retry_path(self) -> None:
-        """重试路径：pending → running → failed → pending → running → completed。"""
-        task = _make_task()
-
-        # 第一次尝试失败
-        task.mark_processing()
-        task.mark_failed("临时错误")
-        assert task.is_failed
-
-        # 重试
-        task.mark_pending_from_failed()
-        assert task.status == GenerationTaskStatus.PENDING
-        assert task.error_message == ""
-
-        # 第二次成功
-        task.mark_processing()
-        task.mark_completed()
-        assert task.is_completed
-
-    def test_cancel_from_pending(self) -> None:
-        """取消路径：pending → cancelled。"""
-        task = _make_task()
-        task.mark_cancelled()
-        assert task.status == GenerationTaskStatus.CANCELLED
-        assert task.is_terminal
-
-    def test_cancel_from_running(self) -> None:
-        """取消路径：pending → running → cancelled。"""
-        task = _make_task()
-        task.mark_processing()
-        task.mark_cancelled()
-        assert task.status == GenerationTaskStatus.CANCELLED
-        assert task.is_terminal