feat: 各コンテナ内のログをポーリング

backend/app/di/container.py

@@ -4,11 +4,14 @@
 
 from app.repository.database import SessionLocal
 from app.repository.setting_repository import SettingRepository
+from app.service.log_stream_service import LogStreamService
 
 
 class DIContainer:
     """依存性注入コンテナ"""
 
+    _log_stream_service: LogStreamService | None = None
+
     @staticmethod
     def get_db() -> Generator[Session]:
         """データベースセッションを取得"""
@@ -21,3 +24,9 @@ def get_db() -> Generator[Session]:
     @staticmethod
     def get_setting_repository(db: Session) -> SettingRepository:
         return SettingRepository(db)
+
+    @classmethod
+    def get_log_stream_service(cls) -> LogStreamService:
+        if cls._log_stream_service is None:
+            cls._log_stream_service = LogStreamService()
+        return cls._log_stream_service

backend/app/di/dependencies.py

@@ -5,6 +5,7 @@
 
 from app.di.container import DIContainer
 from app.repository.setting_repository import SettingRepository
+from app.service.log_stream_service import LogStreamService
 from app.service.s3_service import S3Service
 
 
@@ -19,3 +20,7 @@ def get_setting_repository(db: Session = Depends(get_db)) -> SettingRepository:
 
 def get_s3_service() -> S3Service:
     return S3Service()
+
+
+def get_log_stream_service() -> LogStreamService:
+    return DIContainer.get_log_stream_service()

backend/app/router/sessions.py

@@ -1,12 +1,15 @@
+import asyncio
 import json
+import logging
 from io import BytesIO
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
+from sse_starlette.sse import EventSourceResponse
 
-from app.di.dependencies import get_db, get_s3_service
+from app.di.dependencies import get_db, get_log_stream_service, get_s3_service
 from app.model.session import Iteration, Proposal
 from app.model.session import Session as SessionModel
 from app.schema.session_schema import (
@@ -17,13 +20,16 @@
     ProposalResponse,
     SessionResponse,
 )
+from app.service.log_stream_service import LogStreamService
 from app.service.s3_service import S3Service
 from app.usecase.session_usecase import (
     CreateSessionPRUseCase,
     CreateSessionUseCase,
     IterateUseCase,
 )
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/api/sessions", tags=["sessions"])
 
 
@@ -200,3 +206,28 @@ async def get_diff(
     if not diff:
         raise HTTPException(status_code=404, detail="Diff not found")
     return {"diff": diff}
+
+
+@router.get("/{session_id}/logs/stream")
+async def stream_session_logs(
+    session_id: UUID,
+    since_seconds: int | None = None,
+    log_service: LogStreamService = Depends(get_log_stream_service),
+) -> EventSourceResponse:
+    async def event_generator():  # type: ignore[no-untyped-def]
+        try:
+            async for event in log_service.stream_session_logs(
+                str(session_id), since_seconds=since_seconds
+            ):
+                yield {
+                    "event": "log",
+                    "data": json.dumps(event, ensure_ascii=False),
+                }
+            yield {"event": "done", "data": "{}"}
+        except asyncio.CancelledError:
+            logger.debug("SSE connection closed for session %s", session_id)
+
+    return EventSourceResponse(
+        event_generator(),
+        headers={"X-Accel-Buffering": "no"},
+    )

backend/app/service/k8s_service.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from collections.abc import AsyncIterator
 
 from kubernetes import client, config
 from kubernetes.client.rest import ApiException
@@ -459,6 +460,99 @@ def create_session_pr_job(
         self._create_job_idempotent(job_name, job)
         return job_name
 
+    async def stream_pod_logs(
+        self, job_name: str, tail_lines: int = 100, since_seconds: int | None = None
+    ) -> AsyncIterator[str]:
+        """Stream logs from a job's pod. Yields log lines as they arrive.
+
+        Args:
+            tail_lines: Number of past log lines to include (only used for
+                        pods that are still running and since_seconds is not set).
+            since_seconds: If set, only return logs newer than this many seconds.
+                           Used to avoid replaying old logs on reconnect.
+        """
+        max_wait = 300  # 5 minutes
+        poll_interval = 3
+        elapsed = 0
+
+        # Wait for a pod to appear
+        pod_name: str | None = None
+        sandbox_emitted = False
+        while elapsed < max_wait:
+            try:
+                pods = self.core_v1.list_namespaced_pod(
+                    namespace=self.namespace,
+                    label_selector=f"job-name={job_name}",
+                )
+                if pods.items:
+                    pod = pods.items[0]
+                    pod_name = pod.metadata.name
+                    phase = pod.status.phase if pod.status else None
+                    if phase in ("Running", "Succeeded", "Failed"):
+                        break
+                    # Pod exists but not yet running
+                    if phase == "Pending" and not sandbox_emitted:
+                        yield '@@LOG@@{"phase":"sandbox","message":"Creating sandbox"}'
+                        sandbox_emitted = True
+            except ApiException as e:
+                logger.warning("Error listing pods for job %s: %s", job_name, e)
+            await asyncio.sleep(poll_interval)
+            elapsed += poll_interval
+
+        if not pod_name:
+            msg = f"Pod not found for job {job_name}"
+            yield f'@@LOG@@{{"phase":"waiting","message":"{msg}"}}'
+            return
+
+        # Use a queue to bridge the blocking iterator and async generator
+        queue: asyncio.Queue[str | None] = asyncio.Queue()
+
+        async def _producer() -> None:
+            try:
+                loop = asyncio.get_running_loop()
+
+                kwargs: dict = {
+                    "name": pod_name,
+                    "namespace": self.namespace,
+                    "follow": True,
+                    "_preload_content": False,
+                }
+                if since_seconds is not None:
+                    kwargs["since_seconds"] = since_seconds
+                else:
+                    kwargs["tail_lines"] = tail_lines
+
+                resp = await asyncio.to_thread(
+                    self.core_v1.read_namespaced_pod_log,
+                    **kwargs,
+                )
+
+                def _iter_lines() -> None:
+                    try:
+                        for raw_line in resp:
+                            if isinstance(raw_line, bytes):
+                                line = raw_line.decode("utf-8", errors="replace").rstrip("\n")
+                            else:
+                                line = str(raw_line).rstrip("\n")
+                            loop.call_soon_threadsafe(queue.put_nowait, line)
+                    finally:
+                        loop.call_soon_threadsafe(queue.put_nowait, None)
+
+                await asyncio.to_thread(_iter_lines)
+            except Exception as e:
+                logger.error("Log stream producer error for %s: %s", pod_name, e)
+                await queue.put(None)
+
+        producer_task = asyncio.create_task(_producer())
+        try:
+            while True:
+                line = await queue.get()
+                if line is None:
+                    break
+                yield line
+        finally:
+            producer_task.cancel()
+
     def delete_job(self, job_name: str) -> None:
         """Delete a completed job and its pods."""
         try:

backend/app/service/log_stream_service.py

@@ -0,0 +1,178 @@
+import asyncio
+import json
+import logging
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+
+from app.service.k8s_service import K8sService
+
+logger = logging.getLogger(__name__)
+
+LOG_PREFIX = "@@LOG@@"
+
+
+@dataclass
+class JobInfo:
+    job_name: str
+    job_type: str  # "analyze", "implement", "createpr"
+    proposal_index: int | None = None
+
+
+@dataclass
+class SessionJobs:
+    jobs: list[JobInfo] = field(default_factory=list)
+    version: int = 0  # incremented when jobs list changes
+
+
+def parse_log_line(line: str) -> dict | None:
+    """Parse a log line. If it has the @@LOG@@ prefix, return the parsed JSON.
+    Otherwise return a generic log entry."""
+    if line.startswith(LOG_PREFIX):
+        json_str = line[len(LOG_PREFIX) :]
+        try:
+            result: dict = json.loads(json_str)
+            return result
+        except json.JSONDecodeError:
+            logger.debug("Failed to parse log JSON: %s", json_str[:100])
+            return None
+    return None
+
+
+class LogStreamService:
+    """Manages log streaming for sessions. Singleton per backend process."""
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, SessionJobs] = {}
+        self._lock = asyncio.Lock()
+
+    def register_job(
+        self,
+        session_id: str,
+        job_name: str,
+        job_type: str,
+        proposal_index: int | None = None,
+    ) -> None:
+        """Register a K8s job for log streaming."""
+        if session_id not in self._sessions:
+            self._sessions[session_id] = SessionJobs()
+        session_jobs = self._sessions[session_id]
+        # Avoid duplicate registration
+        for existing in session_jobs.jobs:
+            if existing.job_name == job_name:
+                return
+        session_jobs.jobs.append(
+            JobInfo(job_name=job_name, job_type=job_type, proposal_index=proposal_index)
+        )
+        session_jobs.version += 1
+        logger.info("Registered job %s (type=%s) for session %s", job_name, job_type, session_id)
+
+    def get_session_jobs(self, session_id: str) -> list[JobInfo]:
+        """Get registered jobs for a session."""
+        session_jobs = self._sessions.get(session_id)
+        if not session_jobs:
+            return []
+        return list(session_jobs.jobs)
+
+    async def stream_session_logs(
+        self, session_id: str, since_seconds: int | None = None
+    ) -> AsyncIterator[dict]:
+        """Stream logs from all jobs in a session, multiplexed into a single stream.
+        Dynamically picks up new jobs as they are registered.
+
+        Args:
+            since_seconds: If set, only stream logs newer than this many seconds.
+                           Used on reconnect to avoid replaying old logs.
+        """
+        k8s = K8sService()
+        queue: asyncio.Queue[dict | None] = asyncio.Queue()
+        tracked_jobs: set[str] = set()
+        tasks: list[asyncio.Task[None]] = []
+
+        async def _stream_single_job(job_info: JobInfo) -> None:
+            """Stream logs from a single job into the shared queue."""
+            try:
+                async for line in k8s.stream_pod_logs(
+                    job_info.job_name, since_seconds=since_seconds
+                ):
+                    parsed = parse_log_line(line)
+                    if parsed:
+                        event = {
+                            "job_type": job_info.job_type,
+                            "proposal_index": job_info.proposal_index,
+                            **parsed,
+                        }
+                        await queue.put(event)
+                    else:
+                        # Non-structured log line - emit as raw
+                        await queue.put(
+                            {
+                                "job_type": job_info.job_type,
+                                "proposal_index": job_info.proposal_index,
+                                "phase": "running",
+                                "message": line,
+                            }
+                        )
+            except asyncio.CancelledError:
+                return
+            except Exception as e:
+                logger.error("Error streaming job %s: %s", job_info.job_name, e)
+                await queue.put(
+                    {
+                        "job_type": job_info.job_type,
+                        "proposal_index": job_info.proposal_index,
+                        "phase": "error",
+                        "message": str(e),
+                    }
+                )
+
+        async def _monitor_new_jobs() -> None:
+            """Monitor for new job registrations and start streaming them."""
+            try:
+                while True:
+                    session_jobs = self._sessions.get(session_id)
+                    if session_jobs:
+                        for job_info in session_jobs.jobs:
+                            if job_info.job_name not in tracked_jobs:
+                                tracked_jobs.add(job_info.job_name)
+                                task = asyncio.create_task(_stream_single_job(job_info))
+                                tasks.append(task)
+                    await asyncio.sleep(2)
+            except asyncio.CancelledError:
+                return
+
+        monitor_task = asyncio.create_task(_monitor_new_jobs())
+        tasks.append(monitor_task)
+
+        try:
+            # Keep yielding events until all job streams complete
+            no_event_count = 0
+            while True:
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=5.0)
+                    if event is not None:
+                        no_event_count = 0
+                        yield event
+                except TimeoutError:
+                    no_event_count += 1
+                    # Check if all streaming tasks (excluding monitor) are done
+                    streaming_tasks = [t for t in tasks if t is not monitor_task]
+                    if streaming_tasks and all(t.done() for t in streaming_tasks):
+                        # Drain remaining items
+                        while not queue.empty():
+                            event = queue.get_nowait()
+                            if event is not None:
+                                yield event
+                        break
+                    # After 5 minutes of no events and no active streams, stop
+                    if no_event_count > 60:
+                        break
+                    # Send keepalive
+                    yield {"job_type": "_keepalive", "phase": "waiting", "message": ""}
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    def cleanup_session(self, session_id: str) -> None:
+        """Remove session data after streaming is done."""
+        self._sessions.pop(session_id, None)