Repository navigation
Feat/workspace archive recovery #848
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| # SPDX-FileCopyrightText: 2026 Weibo, Inc. | ||
| # | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| """Internal workspace archive endpoints.""" | ||
|
|
||
| import logging | ||
|
|
||
| from fastapi import APIRouter, Depends, HTTPException | ||
| from pydantic import BaseModel | ||
| from sqlalchemy.orm import Session | ||
|
|
||
| from app.api.dependencies import get_db | ||
| from app.models.subtask import Subtask | ||
| from app.models.task import TaskResource | ||
| from app.schemas.kind import ArchiveInfo | ||
| from app.services.workspace_archive import archive_service | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| router = APIRouter(prefix="/workspace-archives", tags=["internal-workspace-archives"]) | ||
|
|
||
|
|
||
| class ManualArchiveResponse(BaseModel): | ||
| """Response model for manually triggered workspace archive.""" | ||
|
|
||
| success: bool = True | ||
| task_id: int | ||
| archive: ArchiveInfo | ||
|
|
||
|
|
||
| @router.post("/{task_id}/archive", response_model=ManualArchiveResponse) | ||
| async def archive_task_workspace( | ||
| task_id: int, | ||
| db: Session = Depends(get_db), | ||
| ): | ||
| """Archive the current task workspace and persist archive metadata.""" | ||
| task = ( | ||
| db.query(TaskResource) | ||
| .filter( | ||
| TaskResource.id == task_id, | ||
| TaskResource.kind == "Task", | ||
| TaskResource.is_active == TaskResource.STATE_ACTIVE, | ||
| ) | ||
| .first() | ||
| ) | ||
|
|
||
| if not task: | ||
| raise HTTPException(status_code=404, detail="Task not found") | ||
|
|
||
| subtask = ( | ||
| db.query(Subtask) | ||
| .filter( | ||
| Subtask.task_id == task_id, | ||
| Subtask.executor_name.isnot(None), | ||
| Subtask.executor_name != "", | ||
| Subtask.executor_deleted_at == False, | ||
| ) | ||
| .order_by(Subtask.id.desc()) | ||
| .first() | ||
| ) | ||
|
|
||
| if not subtask: | ||
| raise HTTPException(status_code=404, detail="No active executor found for task") | ||
|
|
||
| archive_info = await archive_service.archive_workspace( | ||
| db=db, | ||
| subtask=subtask, | ||
| task=task, | ||
| executor_name=subtask.executor_name, | ||
| executor_namespace=subtask.executor_namespace or "", | ||
| ) | ||
|
|
||
| if not archive_info: | ||
| raise HTTPException(status_code=500, detail="Failed to archive workspace") | ||
|
|
||
| db.commit() | ||
| db.refresh(task) | ||
|
|
||
| logger.info( | ||
| "Manually archived workspace for task %s via executor %s/%s", | ||
| task_id, | ||
| subtask.executor_namespace or "", | ||
| subtask.executor_name, | ||
| ) | ||
|
|
||
| return ManualArchiveResponse(task_id=task_id, archive=archive_info) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,7 +5,7 @@ | |
| import asyncio | ||
| import logging | ||
| from datetime import datetime, timedelta, timezone | ||
| from typing import Dict, List, Set, Tuple | ||
|
|
||
| from sqlalchemy import and_ | ||
| from sqlalchemy.orm import Session | ||
|
|
@@ -131,18 +131,58 @@ def cleanup_stale_executors(self, db: Session) -> None: | |
| ) | ||
| return | ||
|
|
||
| # Deduplicate by (namespace, name) and collect task info for archiving | ||
| # Map: (namespace, name) -> (subtask, task) for archive operation | ||
| executor_task_map: Dict[Tuple[str, str], Tuple[Subtask, TaskResource]] = {} | ||
| for subtask in valid_candidates: | ||
| if subtask.executor_name: | ||
| key = (subtask.executor_namespace, subtask.executor_name) | ||
| if key not in executor_task_map: | ||
| # Get task for this subtask | ||
| task = ( | ||
| db.query(TaskResource) | ||
| .filter( | ||
| TaskResource.id == subtask.task_id, | ||
| TaskResource.kind == "Task", | ||
| TaskResource.is_active == TaskResource.STATE_ACTIVE, | ||
| ) | ||
| .first() | ||
| ) | ||
| if task: | ||
| executor_task_map[key] = (subtask, task) | ||
|
|
||
| if not executor_task_map: | ||
| return | ||
|
|
||
| # Use sync version to avoid event loop issues | ||
| for (ns, name), (subtask, task) in executor_task_map.items(): | ||
| try: | ||
| # Archive workspace before deletion (for code tasks) | ||
| task_crd = Task.model_validate(task.json) | ||
| task_type = ( | ||
| task_crd.metadata.labels | ||
| and task_crd.metadata.labels.get("taskType") | ||
| or "chat" | ||
| ) | ||
|
|
||
| if task_type == "code": | ||
| # Try to archive workspace before deletion | ||
| try: | ||
| self._archive_workspace_sync( | ||
| db=db, | ||
| subtask=subtask, | ||
| task=task, | ||
| executor_name=name, | ||
| executor_namespace=ns, | ||
| ) | ||
| except Exception as archive_error: | ||
| # Log but continue with deletion | ||
| # Recovery will fall back to git clone if archive failed | ||
| logger.warning( | ||
| f"[executor_job] Failed to archive workspace for task {task.id} " | ||
| f"ns={ns} name={name}: {archive_error}" | ||
| ) | ||
|
Comment on lines
+168
to
+184
Contributor
There was a problem hiding this comment. Do not delete code-task executors unless archiving actually succeeded.
Also applies to: 240-258 🧰 Tools🪛 Ruff (0.15.7)[warning] 178-178: Do not catch blind exception: (BLE001) 🤖 Prompt for AI Agents |
||
|
|
||
| logger.info( | ||
| f"[executor_job] Scheduled deleting executor task ns={ns} name={name}" | ||
| ) | ||
|
|
@@ -166,5 +206,59 @@ def cleanup_stale_executors(self, db: Session) -> None: | |
| except Exception as e: | ||
| logger.error(f"[executor_job] cleanup_stale_executors error: {e}") | ||
|
|
||
| def _archive_workspace_sync( | ||
| self, | ||
| db: Session, | ||
| subtask: Subtask, | ||
| task: TaskResource, | ||
| executor_name: str, | ||
| executor_namespace: str, | ||
| ) -> None: | ||
| """Archive workspace files before Pod deletion (synchronous wrapper). | ||
|
|
||
| This method wraps the async archive_service.archive_workspace() for use | ||
| in the synchronous cleanup_stale_executors job. | ||
|
|
||
| Args: | ||
| db: Database session | ||
| subtask: Subtask with executor info | ||
| task: Task resource | ||
| executor_name: Executor name | ||
| executor_namespace: Executor namespace | ||
| """ | ||
| from app.services.workspace_archive import archive_service | ||
|
|
||
| logger.info( | ||
| f"[executor_job] Archiving workspace for task {task.id}, " | ||
| f"executor={executor_namespace}/{executor_name}" | ||
| ) | ||
|
|
||
| # Run async archive in new event loop | ||
| loop = asyncio.new_event_loop() | ||
| asyncio.set_event_loop(loop) | ||
| try: | ||
| archive_info = loop.run_until_complete( | ||
| archive_service.archive_workspace( | ||
| db=db, | ||
| subtask=subtask, | ||
| task=task, | ||
| executor_name=executor_name, | ||
| executor_namespace=executor_namespace, | ||
| ) | ||
| ) | ||
|
|
||
| if archive_info: | ||
| logger.info( | ||
| f"[executor_job] Workspace archived for task {task.id}, " | ||
| f"size={archive_info.sizeBytes} bytes" | ||
| ) | ||
| else: | ||
| logger.info( | ||
| f"[executor_job] Workspace archiving skipped or failed for task {task.id}" | ||
| ) | ||
| finally: | ||
| loop.run_until_complete(loop.shutdown_asyncgens()) | ||
| loop.close() | ||
|
Comment on lines
+237
to
+261
Contributor
There was a problem hiding this comment. 🧩 Analysis chain🏁 Script executed: head -n 300 backend/app/services/adapters/executor_job.py | tail -n 100Repository: wecode-ai/Wegent Length of output: 3985 🏁 Script executed: cd backend && find . -name "*.py" -type f | xargs rg "class JobService" -A 10Repository: wecode-ai/Wegent Length of output: 939 🏁 Script executed: cd backend && rg "job_service\." -B 2 -A 2 | head -40Repository: wecode-ai/Wegent Length of output: 1972 🏁 Script executed: cd backend && cat -n app/services/jobs.py | head -80Repository: wecode-ai/Wegent Length of output: 3029 🏁 Script executed: cd backend && cat -n app/services/jobs.py | sed -n '75,150p'Repository: wecode-ai/Wegent Length of output: 3379 🏁 Script executed: cd backend && cat -n app/services/jobs.py | sed -n '140,160p'Repository: wecode-ai/Wegent Length of output: 1038 🏁 Script executed: cd backend && cat -n app/services/jobs.py | sed -n '159,180p'Repository: wecode-ai/Wegent Length of output: 900 🏁 Script executed: cd backend && sed -n '171,280p' app/services/adapters/executor_job.pyRepository: wecode-ai/Wegent Length of output: 3788 🌐 Web query:
💡 Result: Best practices for asyncio event loop management, especially with background threads in Python: 1. Use high-level entry points: Prefer asyncio.run(coro) or asyncio.Runner for managing the event loop lifecycle. These handle creation, running, cleanup (including shutdown_default_executor and close), and should be used as the main entry point. Avoid manual loop management unless necessary for low-level code. 2. Inside coroutines, use asyncio.get_running_loop instead of deprecated asyncio.get_event_loop. Do not use get_event_loop in async code. 3. For blocking synchronous code (CPU/IO-bound): Offload to thread pool using: - Python 3.9+: asyncio.to_thread(func, *args) for IO-bound (uses default ThreadPoolExecutor). - Or loop.run_in_executor(None, func, *args) or with custom concurrent.futures.ThreadPoolExecutor for more control. Use ProcessPoolExecutor for CPU-bound to bypass GIL. - asyncio.run handles default executor shutdown automatically. Example: import asyncio import concurrent.futures def blocking_io: ... async def main: loop = asyncio.get_running_loop result = await loop.run_in_executor(None, blocking_io) asyncio.run(main) 4. Background threads interacting with asyncio: - Event loop runs in one thread (usually main). asyncio objects are not thread-safe. - From other threads: Use loop.call_soon_threadsafe(callback) for callbacks, asyncio.run_coroutine_threadsafe(coro, loop) to schedule coroutines (returns concurrent.futures.Future). - To run asyncio in background thread: Create new loop with asyncio.new_event_loop, set with asyncio.set_event_loop(loop), then loop.run_forever. Submit tasks via thread-safe methods from main loop thread. Example for per-thread loops: import threading def thread_worker: loop = asyncio.new_event_loop asyncio.set_event_loop(loop) loop.run_forever # Start threads, submit from main via run_coroutine_threadsafe 5. Policies (deprecated in 3.14+, removal 3.16): Avoid set_event_loop_policy. Use loop_factory in asyncio.run(loop_factory=...) for custom loops (e.g., uvloop). 6. Multithreading best practices: - One event loop per thread. - Main thread for signals/subprocesses. - Enable debug mode (asyncio.run(debug=True)) for thread-safety checks, slow callbacks detection. - Close loops explicitly if not using asyncio.run: Call shutdown_asyncgens then close. - No direct coroutine scheduling from other processes (use multiprocessing separately). 7. General: - Avoid blocking the loop: Never run sync blocking code directly in coroutines. - Use create_task(coro) for concurrent async tasks, asyncio.gather for waiting. - For fire-and-forget: Track tasks in set to prevent GC. - Production: Consider uvloop for performance. These practices ensure responsiveness, thread-safety, and proper resource management per official Python 3.14+ docs. Citations:
🏁 Script executed: cd backend && python3 << 'EOF'
# Simulate the proposed fix behavior in a background thread context
import asyncio
def simulate_worker_thread_context():
"""Simulate what happens in cleanup_worker thread"""
print("1. Starting in background thread (no event loop set)")
# Try the proposed fix logic
try:
previous_loop = asyncio.get_event_loop_policy().get_event_loop()
print(f"2. Got previous loop: {previous_loop}")
except RuntimeError as e:
previous_loop = None
print(f"2. RuntimeError caught (expected in worker thread): {e}")
print(" Set previous_loop = None")
# Create new loop (as in _archive_workspace_sync)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
print(f"3. Created and set new loop: {loop}")
# Use and close it
loop.close()
print("4. Closed new loop")
# Try to get event loop after closure
try:
current = asyncio.get_event_loop()
print(f"5. After closure, get_event_loop() returned: {current}")
if current.is_closed():
print(" Loop is closed!")
except RuntimeError as e:
print(f"5. After closure, RuntimeError: {e}")
# Restoration attempt (as in proposed fix)
asyncio.set_event_loop(previous_loop)
print(f"6. Restored previous_loop: {previous_loop}")
simulate_worker_thread_context()
EOFRepository: wecode-ai/Wegent Length of output: 528 🏁 Script executed: cd backend && sed -n '75,95p' app/services/jobs.pyRepository: wecode-ai/Wegent Length of output: 768 Restore the prior event loop after closing it in
Proposed fix- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
+ try:
+ previous_loop = asyncio.get_event_loop_policy().get_event_loop()
+ except RuntimeError:
+ previous_loop = None
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
try:
archive_info = loop.run_until_complete(
archive_service.archive_workspace(
db=db,
subtask=subtask,
@@
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
+ asyncio.set_event_loop(previous_loop)🤖 Prompt for AI Agents |
||
|
|
||
|
|
||
| job_service = JobService(Kind) | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Trace this async internal endpoint.
This new archive control-plane path is missing the repo's async OTel wrapper, so request-level visibility depends on logs only.
As per coding guidelines, "Use
@trace_async(span_name, tracer_name, extract_attributes)decorator to trace entire async functions in OpenTelemetry".🧰 Tools
🪛 Ruff (0.15.7)
[warning] 35-35: Do not perform function call
Dependsin argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable(B008)
🤖 Prompt for AI Agents