feat: プロポーザルにPR関連フィールドを追加し、PR作成機能を実装

backend/app/model/job.py

@@ -61,6 +61,8 @@ class Proposal(Base):
     status = Column(Enum(ProposalStatus), nullable=False, default=ProposalStatus.PENDING)
     after_screenshot_path = Column(String(500), nullable=True)
     diff_path = Column(Text, nullable=True)
+    pr_url = Column(String(500), nullable=True)
+    pr_status = Column(String(20), nullable=True)  # creating, created, failed
     k8s_job_name = Column(String(200), nullable=True)
     error_message = Column(Text, nullable=True)
     created_at = Column(DateTime(timezone=True), default=_utcnow)

backend/app/router/jobs.py

@@ -15,7 +15,7 @@
     ProposalResponse,
 )
 from app.service.artifact_service import ArtifactService
-from app.usecase.job_usecase import CreateJobUseCase, ImplementProposalUseCase
+from app.usecase.job_usecase import CreateJobUseCase, CreatePRUseCase, ImplementProposalUseCase
 
 router = APIRouter(prefix="/api/jobs", tags=["jobs"])
 
@@ -45,6 +45,8 @@ def _to_proposal_response(proposal: Proposal, job_id: UUID) -> ProposalResponse:
             if proposal.after_screenshot_path
             else None
         ),
+        pr_url=proposal.pr_url,
+        pr_status=proposal.pr_status,
         error_message=proposal.error_message,
         created_at=proposal.created_at,
     )
@@ -109,6 +111,20 @@ async def implement_proposals(
     return _to_job_response(job)
 
 
+@router.post("/{job_id}/proposals/{proposal_index}/create-pr", response_model=ProposalResponse)
+async def create_pr(
+    job_id: UUID,
+    proposal_index: int,
+    db: Session = Depends(get_db),
+) -> ProposalResponse:
+    usecase = CreatePRUseCase(db)
+    try:
+        proposal = await usecase.execute(job_id, proposal_index)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    return _to_proposal_response(proposal, job_id)
+
+
 @router.get("/{job_id}/screenshot/before")
 async def get_before_screenshot(
     job_id: UUID, artifacts: ArtifactService = Depends(get_artifact_service)

backend/app/schema/job_schema.py

@@ -22,6 +22,8 @@ class ProposalResponse(BaseModel):
     complexity: str | None
     status: str
     after_screenshot_url: str | None = None
+    pr_url: str | None = None
+    pr_status: str | None = None
     error_message: str | None = None
     created_at: datetime
 

backend/app/service/artifact_service.py

@@ -136,6 +136,17 @@ def get_diff(self, job_id: str, proposal_index: int) -> str | None:
             logger.error("Failed to read diff: %s", e)
             return None
 
+    def get_pr_url(self, job_id: str, proposal_index: int) -> str | None:
+        """Read the PR URL artifact for a proposal."""
+        path = self.base_dir / job_id / "proposals" / str(proposal_index) / "pr_url.txt"
+        if not path.exists():
+            return None
+        try:
+            return path.read_text().strip()
+        except OSError as e:
+            logger.error("Failed to read PR URL: %s", e)
+            return None
+
     def write_proposal_plan(self, job_id: str, proposal_index: int, plan_text: str) -> Path:
         """Write proposal plan to a file (used for local reference)."""
         path = self.base_dir / job_id / f"proposal_{proposal_index}_plan.txt"

backend/app/service/k8s_service.py

@@ -162,6 +162,43 @@ def create_implementation_job(
         logger.info("Created implementation K8s Job: %s", job_name)
         return job_name
 
+    def create_pr_job(
+        self,
+        job_id: str,
+        repo_url: str,
+        branch: str,
+        proposal_index: int,
+    ) -> str:
+        """Create a K8s Job for PR creation. Returns the K8s job name."""
+        job_name = f"ui-worker-{job_id[:8]}-pr-{proposal_index}"
+        labels = {
+            "app": "ui-recommender",
+            "component": "worker",
+            "job-id": job_id[:8],
+            "mode": "createpr",
+        }
+        env_vars = [
+            client.V1EnvVar(name="JOB_ID", value=job_id),
+            client.V1EnvVar(name="REPO_URL", value=repo_url),
+            client.V1EnvVar(name="BRANCH", value=branch),
+            client.V1EnvVar(name="PROPOSAL_INDEX", value=str(proposal_index)),
+            client.V1EnvVar(
+                name="GITHUB_TOKEN",
+                value_from=client.V1EnvVarSource(
+                    secret_key_ref=client.V1SecretKeySelector(
+                        name="ui-recommender-secrets",
+                        key="github-token",
+                    )
+                ),
+            ),
+        ]
+        container = self._build_worker_container("createpr", env_vars)
+        job = self._build_job_spec(job_name, labels, container)
+
+        self.batch_v1.create_namespaced_job(namespace=self.namespace, body=job)
+        logger.info("Created PR creation K8s Job: %s", job_name)
+        return job_name
+
     async def wait_for_job(self, job_name: str, timeout: int = 900, poll_interval: int = 5) -> str:
         """Poll K8s Job status until completion. Returns 'succeeded', 'failed', or 'timeout'."""
         elapsed = 0

backend/app/usecase/job_usecase.py

@@ -12,6 +12,7 @@
 from app.repository.proposal_repository import ProposalRepository
 from app.service.artifact_service import ArtifactService
 from app.workflow.analyzer_graph import build_analyzer_graph
+from app.workflow.create_pr_graph import build_create_pr_graph
 from app.workflow.implementation_graph import build_implementation_graph
 
 logger = logging.getLogger(__name__)
@@ -281,3 +282,118 @@ def _check_job_completion(db: Session, job_id: UUID) -> None:
             new_status = JobStatus.COMPLETED if any_succeeded else JobStatus.FAILED
             job_repo.update_status(job_id, new_status)
             logger.info("Job %s completed with status: %s", job_id, new_status)
+
+
+class CreatePRUseCase:
+    """Trigger PR creation for a completed proposal."""
+
+    def __init__(self, db: Session) -> None:
+        self.db = db
+        self.job_repo = JobRepository(db)
+        self.proposal_repo = ProposalRepository(db)
+
+    async def execute(self, job_id: UUID, proposal_index: int) -> Proposal:
+        job = self.job_repo.get_by_id(job_id)
+        if not job:
+            raise ValueError("Job not found")
+
+        proposal = self.proposal_repo.get_by_job_and_index(job_id, proposal_index)
+        if not proposal:
+            raise ValueError(f"Proposal {proposal_index} not found")
+        if proposal.status != ProposalStatus.COMPLETED:
+            raise ValueError(f"Proposal is not completed: {proposal.status}")
+        if proposal.pr_status == "created":
+            raise ValueError("PR already created")
+        if proposal.pr_status == "creating":
+            raise ValueError("PR creation already in progress")
+
+        proposal_id = UUID(str(proposal.id))
+        self.proposal_repo.update_status(proposal_id, proposal.status, pr_status="creating")
+
+        asyncio.create_task(
+            self._run_create_pr(
+                str(job_id),
+                str(job.repo_url),
+                str(job.branch),
+                proposal_index,
+                str(proposal.id),
+            )
+        )
+
+        updated = self.proposal_repo.get_by_id(proposal_id)
+        if not updated:
+            raise ValueError("Proposal not found after update")
+        return updated
+
+    @staticmethod
+    async def _run_create_pr(
+        job_id: str,
+        repo_url: str,
+        branch: str,
+        proposal_index: int,
+        proposal_id: str,
+    ) -> None:
+        """Background task: run the PR creation LangGraph workflow."""
+        db = SessionLocal()
+        try:
+            proposal_repo = ProposalRepository(db)
+
+            graph = build_create_pr_graph()
+            result = await graph.ainvoke(
+                {
+                    "job_id": job_id,
+                    "repo_url": repo_url,
+                    "branch": branch,
+                    "proposal_index": proposal_index,
+                    "k8s_job_name": None,
+                    "status": "pending",
+                    "error": None,
+                    "pr_url": None,
+                }
+            )
+
+            if result.get("pr_url"):
+                proposal_repo.update_status(
+                    UUID(proposal_id),
+                    ProposalStatus.COMPLETED,
+                    pr_url=result["pr_url"],
+                    pr_status="created",
+                )
+                logger.info(
+                    "PR created for proposal %d of job %s: %s",
+                    proposal_index,
+                    job_id,
+                    result["pr_url"],
+                )
+            else:
+                error = result.get("error", "PR creation failed")
+                proposal_repo.update_status(
+                    UUID(proposal_id),
+                    ProposalStatus.COMPLETED,
+                    pr_status="failed",
+                    error_message=error,
+                )
+                logger.warning(
+                    "PR creation failed for proposal %d of job %s: %s",
+                    proposal_index,
+                    job_id,
+                    error,
+                )
+
+        except Exception:
+            logger.exception(
+                "PR creation failed for proposal %d of job %s",
+                proposal_index,
+                job_id,
+            )
+            try:
+                proposal_repo.update_status(
+                    UUID(proposal_id),
+                    ProposalStatus.COMPLETED,
+                    pr_status="failed",
+                    error_message="Internal error during PR creation",
+                )
+            except Exception:
+                logger.exception("Failed to update proposal PR status")
+        finally:
+            db.close()

backend/app/workflow/create_pr_graph.py

@@ -0,0 +1,81 @@
+import logging
+from typing import Any
+
+from langgraph.graph import END, START, StateGraph
+
+from app.service.artifact_service import ArtifactService
+from app.service.k8s_service import K8sService
+from app.workflow.state import CreatePRState
+
+logger = logging.getLogger(__name__)
+
+
+async def create_k8s_job(state: CreatePRState) -> dict:
+    """Create the K8s Job for PR creation."""
+    k8s = K8sService()
+    job_name = k8s.create_pr_job(
+        job_id=state["job_id"],
+        repo_url=state["repo_url"],
+        branch=state["branch"],
+        proposal_index=state["proposal_index"],
+    )
+    return {"k8s_job_name": job_name, "status": "running"}
+
+
+async def wait_for_job(state: CreatePRState) -> dict:
+    """Poll K8s Job until completion."""
+    k8s = K8sService()
+    k8s_job_name = state["k8s_job_name"]
+    assert k8s_job_name is not None
+    result = await k8s.wait_for_job(k8s_job_name)
+
+    if result == "succeeded":
+        return {"status": "succeeded"}
+
+    logs = k8s.get_job_logs(k8s_job_name)
+    error_msg = f"PR creation job {result}."
+    if logs:
+        error_msg += f" Logs: {logs[:500]}"
+    return {"status": "failed", "error": error_msg}
+
+
+async def extract_results(state: CreatePRState) -> dict:
+    """Extract PR URL artifact from pod logs."""
+    k8s = K8sService()
+    artifacts = ArtifactService()
+
+    k8s_job_name = state["k8s_job_name"]
+    assert k8s_job_name is not None
+    logs = k8s.get_job_logs(k8s_job_name)
+    if logs:
+        artifacts.extract_impl_artifacts_from_logs(state["job_id"], state["proposal_index"], logs)
+
+    pr_url = artifacts.get_pr_url(state["job_id"], state["proposal_index"])
+    return {"pr_url": pr_url}
+
+
+def route_after_wait(state: CreatePRState) -> str:
+    """Route based on job completion status."""
+    if state["status"] == "succeeded":
+        return "extract_results"
+    return END
+
+
+def build_create_pr_graph() -> Any:
+    """Build and compile the PR creation LangGraph."""
+    graph = StateGraph(CreatePRState)
+
+    graph.add_node("create_k8s_job", create_k8s_job)
+    graph.add_node("wait_for_job", wait_for_job)
+    graph.add_node("extract_results", extract_results)
+
+    graph.add_edge(START, "create_k8s_job")
+    graph.add_edge("create_k8s_job", "wait_for_job")
+    graph.add_conditional_edges(
+        "wait_for_job",
+        route_after_wait,
+        {"extract_results": "extract_results", END: END},
+    )
+    graph.add_edge("extract_results", END)
+
+    return graph.compile()

backend/app/workflow/state.py

@@ -39,3 +39,21 @@ class ImplementationState(TypedDict):
     # Output
     after_screenshot_path: str | None
     diff_content: str | None
+
+
+class CreatePRState(TypedDict):
+    """State for the PR creation workflow."""
+
+    # Input
+    job_id: str
+    repo_url: str
+    branch: str
+    proposal_index: int
+
+    # Intermediate
+    k8s_job_name: str | None
+    status: str
+    error: str | None
+
+    # Output
+    pr_url: str | None

backend/migration/versions/a1b2c3d4e5f6_add_pr_fields_to_proposals.py

@@ -0,0 +1,30 @@
+"""add pr fields to proposals
+
+Revision ID: a1b2c3d4e5f6
+Revises: 199f015d6e7d
+Create Date: 2026-02-24 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'a1b2c3d4e5f6'
+down_revision: Union[str, None] = '199f015d6e7d'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add pr_url and pr_status columns to proposals table."""
+    op.add_column('proposals', sa.Column('pr_url', sa.String(length=500), nullable=True))
+    op.add_column('proposals', sa.Column('pr_status', sa.String(length=20), nullable=True))
+
+
+def downgrade() -> None:
+    """Remove pr_url and pr_status columns from proposals table."""
+    op.drop_column('proposals', 'pr_status')
+    op.drop_column('proposals', 'pr_url')