feat(backend): add Completion Condition mechanism for CI monitoring

Add a generic Completion Condition system to track async external
conditions (e.g., CI pipelines) and trigger auto-fix workflows:

- Add CompletionCondition model with status tracking and retry logic
- Add GitHub/GitLab webhook endpoints for receiving CI events
- Add CIMonitorService for handling CI events and auto-fix workflow
- Add CompletionConditionService for CRUD operations
- Add new webhook notification events: condition.satisfied,
  condition.ci_failed, condition.max_retry_reached, task.fully_completed
- Add configuration for CI monitoring (CI_MONITOR_ENABLED, CI_MAX_RETRIES)
- Add database migration for completion_conditions table

The system automatically detects CI failures and creates fix subtasks
with relevant logs, up to a configurable max retry limit.

Author: qdaxb

backend/alembic/versions/2b3c4d5e6f7g_add_completion_conditions_table.py

@@ -0,0 +1,59 @@
+"""add completion_conditions table for CI monitoring
+
+Revision ID: 2b3c4d5e6f7g
+Revises: 1a2b3c4d5e6f
+Create Date: 2025-07-01 12:00:00.000000+08:00
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '2b3c4d5e6f7g'
+down_revision: Union[str, Sequence[str], None] = '1a2b3c4d5e6f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add completion_conditions table for tracking async completion conditions."""
+
+    # Create completion_conditions table
+    op.execute("""
+    CREATE TABLE IF NOT EXISTS completion_conditions (
+        id INT NOT NULL AUTO_INCREMENT,
+        subtask_id INT NOT NULL,
+        task_id INT NOT NULL,
+        user_id INT NOT NULL,
+        condition_type ENUM('CI_PIPELINE', 'EXTERNAL_TASK', 'APPROVAL', 'MANUAL_CONFIRM') NOT NULL DEFAULT 'CI_PIPELINE',
+        status ENUM('PENDING', 'IN_PROGRESS', 'SATISFIED', 'FAILED', 'CANCELLED') NOT NULL DEFAULT 'PENDING',
+        external_id VARCHAR(256) DEFAULT NULL,
+        external_url VARCHAR(1024) DEFAULT NULL,
+        git_platform ENUM('GITHUB', 'GITLAB') DEFAULT NULL,
+        git_domain VARCHAR(256) DEFAULT NULL,
+        repo_full_name VARCHAR(512) DEFAULT NULL,
+        branch_name VARCHAR(256) DEFAULT NULL,
+        retry_count INT NOT NULL DEFAULT 0,
+        max_retries INT NOT NULL DEFAULT 5,
+        last_failure_log TEXT DEFAULT NULL,
+        metadata JSON DEFAULT NULL,
+        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+        satisfied_at DATETIME DEFAULT NULL,
+        PRIMARY KEY (id),
+        KEY ix_completion_conditions_id (id),
+        KEY ix_completion_conditions_subtask_id (subtask_id),
+        KEY ix_completion_conditions_task_id (task_id),
+        KEY ix_completion_conditions_user_id (user_id),
+        KEY ix_completion_conditions_branch_name (branch_name),
+        KEY ix_completion_conditions_repo_branch (repo_full_name, branch_name)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+    """)
+
+
+def downgrade() -> None:
+    """Remove completion_conditions table."""
+    op.execute("DROP TABLE IF EXISTS completion_conditions")

backend/app/api/api.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from app.api.endpoints import admin, auth, oidc, quota, repository, users
+from app.api.endpoints import admin, auth, completion_conditions, oidc, quota, repository, users
 from app.api.endpoints.adapter import (
     agents,
     bots,
@@ -13,6 +13,7 @@
     teams,
 )
 from app.api.endpoints.kind import k_router
+from app.api.endpoints.webhooks import github_router, gitlab_router
 from app.api.router import api_router
 
 api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
@@ -29,3 +30,18 @@
 api_router.include_router(quota.router, prefix="/quota", tags=["quota"])
 api_router.include_router(dify.router, prefix="/dify", tags=["dify"])
 api_router.include_router(k_router)
+
+# Completion conditions and CI monitoring
+api_router.include_router(
+    completion_conditions.router,
+    prefix="/completion-conditions",
+    tags=["completion-conditions"],
+)
+
+# External webhooks (no auth required)
+api_router.include_router(
+    github_router, prefix="/webhooks/github", tags=["webhooks"]
+)
+api_router.include_router(
+    gitlab_router, prefix="/webhooks/gitlab", tags=["webhooks"]
+)

backend/app/api/dependencies.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from contextlib import contextmanager
 from typing import Generator
 
 from sqlalchemy.orm import Session
@@ -19,3 +20,16 @@ def get_db() -> Generator[Session, None, None]:
         yield db
     finally:
         db.close()
+
+
+@contextmanager
+def get_db_context() -> Generator[Session, None, None]:
+    """
+    Database session context manager for use outside of FastAPI dependency injection.
+    Use this when you need a database session in async functions or background tasks.
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/app/api/endpoints/completion_conditions.py

@@ -0,0 +1,141 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Completion Conditions API endpoints
+"""
+from typing import List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+
+from app.api.dependencies import get_db
+from app.core import security
+from app.models.user import User
+from app.schemas.completion_condition import (
+    CompletionConditionCreate,
+    CompletionConditionInDB,
+    CompletionConditionListResponse,
+    TaskCompletionStatus,
+)
+from app.services.completion_condition import completion_condition_service
+
+router = APIRouter()
+
+
+@router.get("", response_model=CompletionConditionListResponse)
+def list_completion_conditions(
+    subtask_id: Optional[int] = Query(None, description="Filter by subtask ID"),
+    task_id: Optional[int] = Query(None, description="Filter by task ID"),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(security.get_current_user),
+):
+    """
+    List completion conditions with optional filters.
+    At least one of subtask_id or task_id must be provided.
+    """
+    if subtask_id is None and task_id is None:
+        raise HTTPException(
+            status_code=400,
+            detail="At least one of subtask_id or task_id must be provided",
+        )
+
+    if subtask_id:
+        conditions = completion_condition_service.get_by_subtask_id(
+            db, subtask_id=subtask_id, user_id=current_user.id
+        )
+    else:
+        conditions = completion_condition_service.get_by_task_id(
+            db, task_id=task_id, user_id=current_user.id
+        )
+
+    return CompletionConditionListResponse(total=len(conditions), items=conditions)
+
+
+@router.get("/{condition_id}", response_model=CompletionConditionInDB)
+def get_completion_condition(
+    condition_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(security.get_current_user),
+):
+    """Get a specific completion condition by ID"""
+    condition = completion_condition_service.get_by_id(
+        db, condition_id=condition_id, user_id=current_user.id
+    )
+    if not condition:
+        raise HTTPException(status_code=404, detail="Completion condition not found")
+    return condition
+
+
+@router.post("", response_model=CompletionConditionInDB)
+def create_completion_condition(
+    condition_in: CompletionConditionCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(security.get_current_user),
+):
+    """Create a new completion condition"""
+    condition = completion_condition_service.create_condition(
+        db, obj_in=condition_in, user_id=current_user.id
+    )
+    return condition
+
+
+@router.delete("/{condition_id}/cancel")
+def cancel_completion_condition(
+    condition_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(security.get_current_user),
+):
+    """Cancel a completion condition"""
+    from app.models.completion_condition import ConditionStatus
+
+    condition = completion_condition_service.get_by_id(
+        db, condition_id=condition_id, user_id=current_user.id
+    )
+    if not condition:
+        raise HTTPException(status_code=404, detail="Completion condition not found")
+
+    if condition.status in [ConditionStatus.SATISFIED, ConditionStatus.FAILED]:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Cannot cancel condition in {condition.status} status",
+        )
+
+    condition = completion_condition_service.update_status(
+        db, condition_id=condition_id, status=ConditionStatus.CANCELLED
+    )
+    return {"status": "cancelled", "id": condition_id}
+
+
+@router.get("/tasks/{task_id}/completion-status", response_model=TaskCompletionStatus)
+def get_task_completion_status(
+    task_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(security.get_current_user),
+):
+    """
+    Get the overall completion status for a task,
+    including all completion conditions and their status.
+    """
+    status = completion_condition_service.get_task_completion_status(
+        db, task_id=task_id, user_id=current_user.id
+    )
+
+    # Convert conditions to schema objects
+    from app.schemas.completion_condition import CompletionConditionInDB
+
+    conditions_in_db = [
+        CompletionConditionInDB.model_validate(c) for c in status["conditions"]
+    ]
+
+    return TaskCompletionStatus(
+        task_id=task_id,
+        subtask_completed=True,  # This would need to be checked from subtask status
+        all_conditions_satisfied=status["all_conditions_satisfied"],
+        pending_conditions=status["pending_conditions"],
+        in_progress_conditions=status["in_progress_conditions"],
+        satisfied_conditions=status["satisfied_conditions"],
+        failed_conditions=status["failed_conditions"],
+        conditions=conditions_in_db,
+    )

backend/app/api/endpoints/webhooks/__init__.py

@@ -0,0 +1,11 @@
+# SPDX-FileCopyrightText: 2025 Weibo, Inc.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Webhooks package for handling external CI events
+"""
+from app.api.endpoints.webhooks.github import router as github_router
+from app.api.endpoints.webhooks.gitlab import router as gitlab_router
+
+__all__ = ["github_router", "gitlab_router"]