diff --git a/codeframe/agents/lead_agent.py b/codeframe/agents/lead_agent.py
index fa5fb614..14495edd 100644
--- a/codeframe/agents/lead_agent.py
+++ b/codeframe/agents/lead_agent.py
@@ -1870,7 +1870,9 @@ async def _assign_and_execute_task(self, task: Task, retry_counts: Dict[int, int
             # Mark agent busy
             self.agent_pool_manager.mark_agent_busy(agent_id, task.id)
 
-            # Update task status to in_progress
+            # Update task with assigned agent and status (Issue #248 fix)
+            # Set assigned_to BEFORE status change so UI shows assignment immediately
+            self.db.update_task(task.id, {"assigned_to": agent_id})
             self.db.update_task(task.id, {"status": "in_progress"})
 
             # Get agent instance
diff --git a/codeframe/ui/routers/tasks.py b/codeframe/ui/routers/tasks.py
index 42586285..515f2f3f 100644
--- a/codeframe/ui/routers/tasks.py
+++ b/codeframe/ui/routers/tasks.py
@@ -370,3 +370,131 @@ async def approve_tasks(
         excluded_count=len(excluded_tasks),
         message=f"Successfully approved {len(approved_tasks)} tasks. Development phase started."
     )
+
+
+# ============================================================================
+# Task Assignment Endpoint (Issue #248 - Manual trigger for stuck tasks)
+# ============================================================================
+
+
+class TaskAssignmentResponse(BaseModel):
+    """Response model for task assignment."""
+    success: bool
+    pending_count: int
+    message: str
+
+
+@project_router.post("/{project_id}/tasks/assign")
+async def assign_pending_tasks(
+    project_id: int,
+    background_tasks: BackgroundTasks,
+    db: Database = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> TaskAssignmentResponse:
+    """Manually trigger task assignment for pending unassigned tasks.
+
+    This endpoint allows users to restart the multi-agent execution process
+    when tasks are stuck in 'pending' state with no agent assigned. This can
+    happen when:
+    - User joins a session after the initial execution completed/failed
+    - The original execution timed out or crashed
+    - WebSocket messages were missed
+
+    Args:
+        project_id: Project ID
+        background_tasks: FastAPI background tasks for async execution
+        db: Database connection
+        current_user: Authenticated user
+
+    Returns:
+        TaskAssignmentResponse with pending task count and status
+
+    Raises:
+        HTTPException:
+            - 400: Project not in active phase
+            - 403: Access denied
+            - 404: Project not found
+    """
+    # Verify project exists
+    project = db.get_project(project_id)
+    if not project:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Project {project_id} not found"
+        )
+
+    # Authorization check
+    if not db.user_has_project_access(current_user.id, project_id):
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    # Validate project is in active phase (development)
+    current_phase = project.get("phase", "discovery")
+    if current_phase != "active":
+        raise HTTPException(
+            status_code=400,
+            detail=f"Project must be in active (development) phase to assign tasks. Current phase: {current_phase}"
+        )
+
+    # Get all tasks and count pending unassigned ones
+    tasks = db.get_project_tasks(project_id)
+    pending_unassigned = [
+        t for t in tasks
+        if t.status == TaskStatus.PENDING and not t.assigned_to
+    ]
+    pending_count = len(pending_unassigned)
+
+    if pending_count == 0:
+        # Debug logging to help diagnose why tasks might appear stuck
+        logger.debug(
+            f"assign_pending_tasks called for project {project_id} but found 0 pending unassigned tasks. "
+            f"Total tasks: {len(tasks)}, statuses: {[t.status.value for t in tasks]}"
+        )
+        return TaskAssignmentResponse(
+            success=True,
+            pending_count=0,
+            message="No pending unassigned tasks to assign."
+        )
+
+    # Check if execution is already in progress (Phase 1 fix for concurrent execution)
+    # Include ASSIGNED status to prevent race between assignment and execution start
+    executing_tasks = [
+        t for t in tasks
+        if t.status in [TaskStatus.ASSIGNED, TaskStatus.IN_PROGRESS]
+    ]
+    if executing_tasks:
+        logger.info(
+            f"⏳ Execution already in progress for project {project_id}: "
+            f"{len(executing_tasks)} tasks assigned/running"
+        )
+        return TaskAssignmentResponse(
+            success=True,
+            pending_count=pending_count,
+            message=f"Execution already in progress ({len(executing_tasks)} task(s) assigned/running). Please wait."
+        )
+
+    # Schedule multi-agent execution in background
+    api_key = os.environ.get("ANTHROPIC_API_KEY")
+    if not api_key:
+        logger.warning(
+            f"⚠️  ANTHROPIC_API_KEY not configured - cannot assign tasks for project {project_id}"
+        )
+        return TaskAssignmentResponse(
+            success=False,
+            pending_count=pending_count,
+            message="Cannot assign tasks: API key not configured. Please contact administrator."
+        )
+
+    background_tasks.add_task(
+        start_development_execution,
+        project_id,
+        db,
+        manager,
+        api_key
+    )
+    logger.info(f"✅ Scheduled task assignment for project {project_id} ({pending_count} pending tasks)")
+
+    return TaskAssignmentResponse(
+        success=True,
+        pending_count=pending_count,
+        message=f"Assignment started for {pending_count} pending task(s)."
+    )
diff --git a/tests/integration/test_multi_agent_execution.py b/tests/integration/test_multi_agent_execution.py
index dedb7f6d..89552abc 100644
--- a/tests/integration/test_multi_agent_execution.py
+++ b/tests/integration/test_multi_agent_execution.py
@@ -667,3 +667,229 @@ def record_usage(i: int):
         total = cursor.fetchone()["total"]
         expected_total = sum(100 + i for i in range(20))
         assert total == expected_total
+
+
+@pytest.mark.integration
+class TestAssignAndExecuteTaskAssignedTo:
+    """Tests for _assign_and_execute_task setting assigned_to field (Issue #248 fix)."""
+
+    @pytest.mark.asyncio
+    async def test_assign_and_execute_task_sets_assigned_to(
+        self, real_db: Database, test_workspace: Path
+    ):
+        """Test that _assign_and_execute_task sets the assigned_to field on the task.
+
+        This is a regression test for Issue #248 where tasks remained showing
+        'Assigned to: Unassigned' because the assigned_to field was never populated
+        during task execution.
+        """
+        from codeframe.agents.lead_agent import LeadAgent
+
+        # Setup project
+        project_id = real_db.create_project(
+            name="assigned-to-test",
+            description="Test assigned_to field population",
+            source_type="empty",
+            workspace_path=str(test_workspace),
+        )
+
+        # Create issue and task
+        issue_id = real_db.create_issue({
+            "project_id": project_id,
+            "issue_number": "AT-001",
+            "title": "Test Issue",
+            "description": "Test issue for assigned_to",
+            "priority": 1,
+            "workflow_step": 1,
+        })
+
+        task_id = real_db.create_task_with_issue(
+            project_id=project_id,
+            issue_id=issue_id,
+            task_number="AT-001-1",
+            parent_issue_number="AT-001",
+            title="Test Task for Assignment",
+            description="This task should have assigned_to set",
+            status=TaskStatus.PENDING,
+            priority=1,
+            workflow_step=1,
+            can_parallelize=True,
+        )
+
+        # Verify task starts with no assigned_to
+        task_before = real_db.get_task(task_id)
+        assert task_before.assigned_to is None, "Task should start unassigned"
+
+        # Create LeadAgent with mocked execution
+        with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "sk-ant-test-key"}):
+            with patch("codeframe.agents.lead_agent.AgentPoolManager") as mock_pool_class:
+                # Setup mock pool manager
+                mock_pool = Mock()
+                mock_pool.get_or_create_agent.return_value = "test-agent-001"
+                mock_pool.mark_agent_busy.return_value = None
+                mock_pool.mark_agent_idle.return_value = None
+                mock_pool.get_agent_status.return_value = {
+                    "test-agent-001": {"status": "idle", "agent_type": "backend"}
+                }
+
+                # Mock agent instance with execute_task
+                mock_agent_instance = Mock()
+                mock_agent_instance.execute_task = AsyncMock(return_value={"status": "completed"})
+                mock_pool.get_agent_instance.return_value = mock_agent_instance
+
+                mock_pool_class.return_value = mock_pool
+
+                lead_agent = LeadAgent(
+                    project_id=project_id,
+                    db=real_db,
+                    api_key="sk-ant-test-key",
+                    ws_manager=None,
+                )
+                lead_agent.agent_pool_manager = mock_pool
+
+                # Also mock the review agent to avoid review step
+                with patch.object(lead_agent.agent_pool_manager, "get_or_create_agent") as mock_get_agent:
+                    # First call returns worker agent, second call returns review agent
+                    mock_get_agent.side_effect = ["test-agent-001", "review-agent-001"]
+
+                    mock_review_instance = Mock()
+                    mock_review_report = Mock()
+                    mock_review_report.status = "approved"
+                    mock_review_report.overall_score = 9.0
+                    mock_review_instance.execute_task = AsyncMock(return_value=mock_review_report)
+
+                    def get_instance_side_effect(agent_id):
+                        if agent_id == "review-agent-001":
+                            return mock_review_instance
+                        return mock_agent_instance
+
+                    mock_pool.get_agent_instance.side_effect = get_instance_side_effect
+
+                    # Get the task object
+                    task = real_db.get_task(task_id)
+
+                    # Execute _assign_and_execute_task
+                    retry_counts = {}
+                    result = await lead_agent._assign_and_execute_task(task, retry_counts)
+
+                    assert result is True, "Task execution should succeed"
+
+        # CRITICAL ASSERTION: Verify assigned_to was set
+        task_after = real_db.get_task(task_id)
+        assert task_after.assigned_to == "test-agent-001", (
+            f"Task assigned_to should be 'test-agent-001' but was '{task_after.assigned_to}'. "
+            "This is the Issue #248 bug - assigned_to field not being populated."
+        )
+
+    @pytest.mark.asyncio
+    async def test_assign_and_execute_task_sets_assigned_to_before_in_progress(
+        self, real_db: Database, test_workspace: Path
+    ):
+        """Test that assigned_to is set before status changes to in_progress.
+
+        The UI needs to show assignment even during the brief period before
+        task execution begins.
+        """
+        from codeframe.agents.lead_agent import LeadAgent
+
+        # Setup project
+        project_id = real_db.create_project(
+            name="assigned-to-order-test",
+            description="Test assigned_to ordering",
+            source_type="empty",
+            workspace_path=str(test_workspace),
+        )
+
+        issue_id = real_db.create_issue({
+            "project_id": project_id,
+            "issue_number": "AO-001",
+            "title": "Order Test Issue",
+            "description": "Test",
+            "priority": 1,
+            "workflow_step": 1,
+        })
+
+        task_id = real_db.create_task_with_issue(
+            project_id=project_id,
+            issue_id=issue_id,
+            task_number="AO-001-1",
+            parent_issue_number="AO-001",
+            title="Order Test Task",
+            description="Test ordering",
+            status=TaskStatus.PENDING,
+            priority=1,
+            workflow_step=1,
+            can_parallelize=True,
+        )
+
+        # Track database update calls to verify ordering
+        update_calls = []
+        original_update_task = real_db.update_task
+
+        def tracking_update_task(task_id, updates):
+            update_calls.append((task_id, updates.copy()))
+            return original_update_task(task_id, updates)
+
+        with patch.dict(os.environ, {"ANTHROPIC_API_KEY": "sk-ant-test-key"}):
+            with patch("codeframe.agents.lead_agent.AgentPoolManager") as mock_pool_class:
+                mock_pool = Mock()
+                mock_pool.get_or_create_agent.return_value = "order-agent-001"
+                mock_pool.mark_agent_busy.return_value = None
+                mock_pool.mark_agent_idle.return_value = None
+                mock_pool.get_agent_status.return_value = {}
+
+                mock_agent_instance = Mock()
+                mock_agent_instance.execute_task = AsyncMock(return_value={"status": "completed"})
+
+                mock_review_instance = Mock()
+                mock_review_report = Mock()
+                mock_review_report.status = "approved"
+                mock_review_report.overall_score = 9.0
+                mock_review_instance.execute_task = AsyncMock(return_value=mock_review_report)
+
+                def get_agent_side_effect(agent_type):
+                    if agent_type == "review":
+                        return "review-agent-001"
+                    return "order-agent-001"
+
+                mock_pool.get_or_create_agent.side_effect = get_agent_side_effect
+
+                def get_instance_side_effect(agent_id):
+                    if agent_id == "review-agent-001":
+                        return mock_review_instance
+                    return mock_agent_instance
+
+                mock_pool.get_agent_instance.side_effect = get_instance_side_effect
+
+                mock_pool_class.return_value = mock_pool
+
+                lead_agent = LeadAgent(
+                    project_id=project_id,
+                    db=real_db,
+                    api_key="sk-ant-test-key",
+                    ws_manager=None,
+                )
+                lead_agent.agent_pool_manager = mock_pool
+
+                # Patch update_task to track calls
+                with patch.object(real_db, "update_task", side_effect=tracking_update_task):
+                    task = real_db.get_task(task_id)
+                    retry_counts = {}
+                    await lead_agent._assign_and_execute_task(task, retry_counts)
+
+        # Verify update order: assigned_to should be set before or with in_progress
+        assigned_to_index = None
+        in_progress_index = None
+
+        for i, (tid, updates) in enumerate(update_calls):
+            if "assigned_to" in updates:
+                assigned_to_index = i
+            if updates.get("status") == "in_progress":
+                in_progress_index = i
+
+        assert assigned_to_index is not None, "assigned_to should be updated"
+        assert in_progress_index is not None, "status should be updated to in_progress"
+        assert assigned_to_index <= in_progress_index, (
+            f"assigned_to (call {assigned_to_index}) should be set before or with "
+            f"in_progress (call {in_progress_index})"
+        )
diff --git a/tests/ui/test_assign_pending_tasks.py b/tests/ui/test_assign_pending_tasks.py
new file mode 100644
index 00000000..9a8f4cfa
--- /dev/null
+++ b/tests/ui/test_assign_pending_tasks.py
@@ -0,0 +1,369 @@
+"""
+Tests for assign pending tasks endpoint (Issue #248 fix).
+
+This module tests the POST /api/projects/{project_id}/tasks/assign endpoint
+that allows users to manually trigger task assignment for stuck pending tasks.
+"""
+
+import os
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi import BackgroundTasks, HTTPException
+
+from codeframe.core.models import Task, TaskStatus
+
+
+@pytest.fixture
+def mock_background_tasks(monkeypatch):
+    """Create mock BackgroundTasks.
+
+    Also clears ANTHROPIC_API_KEY to make tests deterministic.
+    Tests that need API key behavior should explicitly set it.
+    """
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    bg = MagicMock(spec=BackgroundTasks)
+    bg.add_task = MagicMock()
+    return bg
+
+
+@pytest.fixture
+def mock_db():
+    """Create mock Database with project in active phase."""
+    db = MagicMock()
+    db.get_project.return_value = {
+        "id": 1,
+        "name": "Test Project",
+        "phase": "active"  # Must be in active phase to assign tasks
+    }
+    db.user_has_project_access.return_value = True
+    return db
+
+
+@pytest.fixture
+def mock_user():
+    """Create mock authenticated user."""
+    user = MagicMock()
+    user.id = 1
+    user.email = "test@example.com"
+    return user
+
+
+@pytest.fixture
+def mock_manager():
+    """Create mock ConnectionManager."""
+    manager = MagicMock()
+    manager.broadcast = AsyncMock()
+    return manager
+
+
+class TestAssignPendingTasksEndpoint:
+    """Tests for POST /api/projects/{project_id}/tasks/assign."""
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_with_pending_tasks(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that endpoint triggers execution when pending tasks exist."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: 2 pending unassigned tasks
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=3, project_id=1, title="Task 3", status=TaskStatus.COMPLETED, assigned_to="agent-1"),
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        assert response.success is True
+        assert response.pending_count == 2
+        assert "2" in response.message
+        mock_background_tasks.add_task.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_no_pending_tasks(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that endpoint returns success but doesn't trigger execution when no pending tasks."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: No pending tasks
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.COMPLETED, assigned_to="agent-1"),
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.IN_PROGRESS, assigned_to="agent-2"),
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        assert response.success is True
+        assert response.pending_count == 0
+        assert "no pending" in response.message.lower()
+        mock_background_tasks.add_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_wrong_phase(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that endpoint returns 400 when project is not in active phase."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: Project in planning phase
+        mock_db.get_project.return_value = {
+            "id": 1,
+            "name": "Test Project",
+            "phase": "planning"
+        }
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             pytest.raises(HTTPException) as exc_info:
+            await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        assert exc_info.value.status_code == 400
+        assert "active" in exc_info.value.detail.lower()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_project_not_found(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that endpoint returns 404 when project doesn't exist."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        mock_db.get_project.return_value = None
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             pytest.raises(HTTPException) as exc_info:
+            await assign_pending_tasks(
+                project_id=999,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        assert exc_info.value.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_access_denied(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that endpoint returns 403 when user doesn't have access."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        mock_db.user_has_project_access.return_value = False
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             pytest.raises(HTTPException) as exc_info:
+            await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        assert exc_info.value.status_code == 403
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_without_api_key(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks, caplog
+    ):
+        """Test that endpoint warns when API key is missing."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+        import logging
+
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),
+        ]
+
+        # Ensure no API key
+        env_without_key = {k: v for k, v in os.environ.items() if k != "ANTHROPIC_API_KEY"}
+
+        with caplog.at_level(logging.WARNING):
+            with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+                 patch.dict(os.environ, env_without_key, clear=True):
+                response = await assign_pending_tasks(
+                    project_id=1,
+                    background_tasks=mock_background_tasks,
+                    db=mock_db,
+                    current_user=mock_user
+                )
+
+        # Should return failure when API key missing
+        assert response.success is False
+        assert response.pending_count == 1
+        assert "api key" in response.message.lower() or "not configured" in response.message.lower()
+
+        # Background task should NOT be scheduled
+        mock_background_tasks.add_task.assert_not_called()
+
+        # Should log warning
+        assert any("ANTHROPIC_API_KEY" in record.message for record in caplog.records)
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_only_counts_unassigned(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that only pending AND unassigned tasks are counted."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: Mix of tasks - only 1 is pending AND unassigned
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),      # Count this
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.PENDING, assigned_to="agent-1"), # Already assigned
+            Task(id=3, project_id=1, title="Task 3", status=TaskStatus.IN_PROGRESS, assigned_to=None),  # Not pending
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        # Should not trigger because there's an in_progress task
+        assert response.pending_count == 1
+        assert "in progress" in response.message.lower()
+        mock_background_tasks.add_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_blocked_when_execution_in_progress(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that assignment is blocked when tasks are already in progress."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: 2 pending tasks + 1 in progress
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=3, project_id=1, title="Task 3", status=TaskStatus.IN_PROGRESS, assigned_to="agent-1"),
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        # Should return success but NOT schedule execution
+        assert response.success is True
+        assert response.pending_count == 2
+        assert "in progress" in response.message.lower()
+        assert "1 task" in response.message.lower()  # Reports 1 task running
+        mock_background_tasks.add_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_allowed_when_no_execution_in_progress(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that assignment proceeds when no tasks are in progress."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: Only pending and completed tasks, no in_progress or assigned
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=3, project_id=1, title="Task 3", status=TaskStatus.COMPLETED, assigned_to="agent-1"),
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        # Should schedule execution
+        assert response.success is True
+        assert response.pending_count == 2
+        assert "started" in response.message.lower()
+        mock_background_tasks.add_task.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_assign_pending_tasks_blocked_when_tasks_assigned(
+        self, mock_db, mock_user, mock_manager, mock_background_tasks
+    ):
+        """Test that assignment is blocked when tasks are in ASSIGNED status."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+
+        # Setup: 2 pending tasks + 1 assigned (not yet in_progress)
+        mock_db.get_project_tasks.return_value = [
+            Task(id=1, project_id=1, title="Task 1", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=2, project_id=1, title="Task 2", status=TaskStatus.PENDING, assigned_to=None),
+            Task(id=3, project_id=1, title="Task 3", status=TaskStatus.ASSIGNED, assigned_to="agent-1"),
+        ]
+
+        with patch("codeframe.ui.routers.tasks.manager", mock_manager), \
+             patch.dict(os.environ, {"ANTHROPIC_API_KEY": "test-key"}):
+            response = await assign_pending_tasks(
+                project_id=1,
+                background_tasks=mock_background_tasks,
+                db=mock_db,
+                current_user=mock_user
+            )
+
+        # Should return success but NOT schedule execution
+        assert response.success is True
+        assert response.pending_count == 2
+        assert "in progress" in response.message.lower() or "assigned" in response.message.lower()
+        mock_background_tasks.add_task.assert_not_called()
+
+
+class TestAssignPendingTasksResponseModel:
+    """Tests for TaskAssignmentResponse model."""
+
+    def test_response_model_structure(self):
+        """Test that response model has correct fields."""
+        from codeframe.ui.routers.tasks import TaskAssignmentResponse
+
+        response = TaskAssignmentResponse(
+            success=True,
+            pending_count=5,
+            message="Test message"
+        )
+
+        assert response.success is True
+        assert response.pending_count == 5
+        assert response.message == "Test message"
+
+
+class TestAssignPendingTasksFunctionSignature:
+    """Tests to ensure endpoint signature is correct."""
+
+    def test_endpoint_accepts_required_parameters(self):
+        """Test that endpoint function accepts required parameters."""
+        from codeframe.ui.routers.tasks import assign_pending_tasks
+        import inspect
+
+        sig = inspect.signature(assign_pending_tasks)
+        params = list(sig.parameters.keys())
+
+        assert "project_id" in params
+        assert "background_tasks" in params
+        assert "db" in params
+        assert "current_user" in params
diff --git a/web-ui/src/components/TaskList.tsx b/web-ui/src/components/TaskList.tsx
index d95f47bc..d6703a02 100644
--- a/web-ui/src/components/TaskList.tsx
+++ b/web-ui/src/components/TaskList.tsx
@@ -11,6 +11,7 @@
 
 import { useState, useMemo, useCallback, memo } from 'react';
 import { useAgentState } from '@/hooks/useAgentState';
+import { tasksApi } from '@/lib/api';
 import QualityGateStatus from '@/components/quality-gates/QualityGateStatus';
 import ErrorBoundary from '@/components/ErrorBoundary';
 import type { Task, TaskStatus } from '@/types/agentState';
@@ -178,6 +179,10 @@ const TaskList = memo(function TaskList({ projectId }: TaskListProps) {
   // Quality gates visibility state (keyed by task ID)
   const [qualityGatesVisible, setQualityGatesVisible] = useState<Set<number>>(new Set());
 
+  // Assignment state (Issue #248 fix)
+  const [isAssigning, setIsAssigning] = useState(false);
+  const [assignmentError, setAssignmentError] = useState<string | null>(null);
+
   // Filter tasks by project ID
   const projectTasks = useMemo(
     () => tasks.filter((task) => task.project_id === projectId),
@@ -213,6 +218,43 @@ const TaskList = memo(function TaskList({ projectId }: TaskListProps) {
     return projectTasks.filter((task) => task.status === activeFilter);
   }, [projectTasks, activeFilter]);
 
+  // Check if there are pending unassigned tasks (Issue #248 fix)
+  const hasPendingUnassigned = useMemo(() => {
+    return projectTasks.some(
+      (task) => task.status === 'pending' && !task.agent_id
+    );
+  }, [projectTasks]);
+
+  // Count of pending unassigned tasks
+  const pendingUnassignedCount = useMemo(() => {
+    return projectTasks.filter(
+      (task) => task.status === 'pending' && !task.agent_id
+    ).length;
+  }, [projectTasks]);
+
+  // Check if execution is already in progress (Phase 1 fix for concurrent execution)
+  const hasTasksInProgress = useMemo(() => {
+    return projectTasks.some((task) => task.status === 'in_progress');
+  }, [projectTasks]);
+
+  // Handler for Assign Tasks button (Issue #248 fix)
+  const handleAssignTasks = useCallback(async () => {
+    setIsAssigning(true);
+    setAssignmentError(null);
+    try {
+      const response = await tasksApi.assignPending(projectId);
+      if (!response.data.success) {
+        setAssignmentError(response.data.message || 'Assignment failed');
+      }
+      // Success - the WebSocket will update the task list as agents are assigned
+    } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : 'Failed to assign tasks';
+      setAssignmentError(errorMessage);
+    } finally {
+      setIsAssigning(false);
+    }
+  }, [projectId]);
+
   // Toggle quality gates visibility for a task
   const handleViewQualityGates = useCallback((taskId: number) => {
     setQualityGatesVisible((prev) => {
@@ -261,6 +303,38 @@ const TaskList = memo(function TaskList({ projectId }: TaskListProps) {
         </span>
       </div>
 
+      {/* Pending Unassigned Tasks Banner (Issue #248 fix) */}
+      {hasPendingUnassigned && (
+        <div
+          data-testid="assign-tasks-banner"
+          className="mb-4 p-4 bg-muted/50 border border-border rounded-lg"
+        >
+          <div className="flex items-center justify-between gap-4">
+            <div className="flex-1">
+              <h3 className="font-medium text-foreground">Tasks Pending Assignment</h3>
+              <p className="text-sm text-muted-foreground">
+                {hasTasksInProgress ? (
+                  <>Execution in progress. {pendingUnassignedCount} task{pendingUnassignedCount !== 1 ? 's' : ''} waiting...</>
+                ) : (
+                  <>{pendingUnassignedCount} task{pendingUnassignedCount !== 1 ? 's are' : ' is'} waiting to be assigned to agents</>
+                )}
+              </p>
+            </div>
+            <button
+              data-testid="assign-tasks-button"
+              onClick={handleAssignTasks}
+              disabled={isAssigning || hasTasksInProgress}
+              className="px-4 py-2 bg-primary text-primary-foreground rounded-md hover:bg-primary/90 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
+            >
+              {isAssigning ? 'Assigning...' : hasTasksInProgress ? 'Running...' : 'Assign Tasks'}
+            </button>
+          </div>
+          {assignmentError && (
+            <p className="mt-2 text-sm text-destructive">{assignmentError}</p>
+          )}
+        </div>
+      )}
+
       {/* Filter Buttons */}
       <div className="flex flex-wrap gap-2 mb-4" role="group" aria-label="Filter tasks by status">
         {FILTER_OPTIONS.map((option) => (
diff --git a/web-ui/src/lib/api.ts b/web-ui/src/lib/api.ts
index 784d5869..1d14ac97 100644
--- a/web-ui/src/lib/api.ts
+++ b/web-ui/src/lib/api.ts
@@ -150,6 +150,21 @@ export const tasksApi = {
     api.get<{ tasks: Task[]; total: number }>(`/api/projects/${projectId}/tasks`, {
       params: filters,
     }),
+  /**
+   * Manually trigger assignment for pending unassigned tasks (Issue #248 fix).
+   *
+   * This endpoint allows users to restart the multi-agent execution process
+   * when tasks are stuck in 'pending' state with no agent assigned.
+   *
+   * @param projectId - Project ID
+   * @returns Assignment response with pending count and status message
+   */
+  assignPending: (projectId: number) =>
+    api.post<{
+      success: boolean;
+      pending_count: number;
+      message: string;
+    }>(`/api/projects/${projectId}/tasks/assign`),
 };
 
 export const blockersApi = {