From 32c31dedf58f0f24085d5ed63a9eae157d4a96a0 Mon Sep 17 00:00:00 2001
From: yu-sugimoto <yuyuyuk2002@gmail.com>
Date: Wed, 4 Mar 2026 11:07:14 +0900
Subject: [PATCH 1/5] =?UTF-8?q?feat:=20=E3=83=AD=E3=82=B0=E3=82=B7?=
 =?UTF-8?q?=E3=82=B9=E3=83=86=E3=83=A0=E3=81=AE=E6=A7=8B=E7=AF=89=E3=81=A8?=
 =?UTF-8?q?=E3=82=A8=E3=83=BC=E3=82=B8=E3=82=A7=E3=83=B3=E3=83=88=E7=9B=A3?=
 =?UTF-8?q?=E8=A6=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/app/di/container.py                   |   9 +
 backend/app/di/dependencies.py                |   5 +
 backend/app/router/sessions.py                |  33 ++-
 backend/app/service/k8s_service.py            |  88 ++++++
 backend/app/service/log_stream_service.py     | 175 +++++++++++
 backend/app/usecase/session_usecase.py        |  88 +++---
 .../app/workflow/session_analyzer_graph.py    |   9 +
 .../app/workflow/session_create_pr_graph.py   |  10 +
 .../workflow/session_implementation_graph.py  |  10 +
 ...d4e5f6g7h8_add_session_iteration_tables.py |  49 ++--
 backend/pyproject.toml                        |   3 +-
 backend/uv.lock                               |  15 +
 docker/worker-analyze.py                      | 146 ++++++++--
 docker/worker-createpr.py                     |  81 +++++-
 docker/worker-implement.py                    |  90 +++++-
 frontend/src/components/LogPanel.tsx          | 275 ++++++++++++++++++
 frontend/src/hooks/useLogStream.ts            | 150 ++++++++++
 frontend/src/pages/SessionDetail.tsx          |  33 +--
 18 files changed, 1138 insertions(+), 131 deletions(-)
 create mode 100644 backend/app/service/log_stream_service.py
 create mode 100644 frontend/src/components/LogPanel.tsx
 create mode 100644 frontend/src/hooks/useLogStream.ts

diff --git a/backend/app/di/container.py b/backend/app/di/container.py
index 2c27830..1a71db8 100644
--- a/backend/app/di/container.py
+++ b/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
diff --git a/backend/app/di/dependencies.py b/backend/app/di/dependencies.py
index 1800329..2e838dc 100644
--- a/backend/app/di/dependencies.py
+++ b/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()
diff --git a/backend/app/router/sessions.py b/backend/app/router/sessions.py
index 8d5f3f2..71531f0 100644
--- a/backend/app/router/sessions.py
+++ b/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,6 +20,7 @@
     ProposalResponse,
     SessionResponse,
 )
+from app.service.log_stream_service import LogStreamService
 from app.service.s3_service import S3Service
 from app.usecase.session_usecase import (
     CreateSessionPRUseCase,
@@ -24,6 +28,8 @@
     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"},
+    )
diff --git a/backend/app/service/k8s_service.py b/backend/app/service/k8s_service.py
index 6fc2d30..d786873 100644
--- a/backend/app/service/k8s_service.py
+++ b/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,93 @@ 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
+        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
+            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:
+            yield f"@@LOG@@{{\"phase\":\"waiting\",\"message\":\"Pod not found for job {job_name}\"}}"
+            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:
diff --git a/backend/app/service/log_stream_service.py b/backend/app/service/log_stream_service.py
new file mode 100644
index 0000000..3785929
--- /dev/null
+++ b/backend/app/service/log_stream_service.py
@@ -0,0 +1,175 @@
+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:
+            return json.loads(json_str)
+        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] = []  # type: ignore[type-arg]
+
+        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)
diff --git a/backend/app/usecase/session_usecase.py b/backend/app/usecase/session_usecase.py
index f53268a..cb5aae8 100644
--- a/backend/app/usecase/session_usecase.py
+++ b/backend/app/usecase/session_usecase.py
@@ -27,6 +27,38 @@
 logger = logging.getLogger(__name__)
 
 
+def _update_iteration_status_with_retry(
+    iter_repo: IterationRepository,
+    iteration_id: UUID,
+    new_status: IterationStatus,
+    max_retries: int = 3,
+    **kwargs: object,
+) -> Iteration | None:
+    """Update iteration status with retry on version mismatch."""
+    for attempt in range(max_retries):
+        iteration = iter_repo.get_by_id(iteration_id)
+        if not iteration:
+            return None
+        result = iter_repo.update_status_optimistic(
+            iteration_id, iteration.version, new_status, **kwargs
+        )
+        if result:
+            return result
+        logger.warning(
+            "Iteration %s status update to %s failed (attempt %d), retrying...",
+            iteration_id,
+            new_status,
+            attempt + 1,
+        )
+    logger.error(
+        "Failed to update iteration %s to %s after %d retries",
+        iteration_id,
+        new_status,
+        max_retries,
+    )
+    return None
+
+
 class CreateSessionUseCase:
     """Create a new session and trigger the first analysis."""
 
@@ -231,8 +263,8 @@ async def _run_session_analysis(
             logger.error("Iteration %s not found", iteration_id)
             return
 
-        iter_repo.update_status_optimistic(
-            UUID(iteration_id), iteration.version, IterationStatus.ANALYZING
+        _update_iteration_status_with_retry(
+            iter_repo, UUID(iteration_id), IterationStatus.ANALYZING
         )
 
         graph = build_session_analyzer_graph()
@@ -254,11 +286,6 @@ async def _run_session_analysis(
         )
 
         if result.get("proposals"):
-            # Refresh iteration for current version
-            iteration = iter_repo.get_by_id(UUID(iteration_id))
-            if not iteration:
-                return
-
             proposals = []
             for i, prop in enumerate(result["proposals"]):
                 p = Proposal(
@@ -273,9 +300,9 @@ async def _run_session_analysis(
                 )
                 proposals.append(proposal_repo.create(p))
 
-            iter_repo.update_status_optimistic(
+            _update_iteration_status_with_retry(
+                iter_repo,
                 UUID(iteration_id),
-                iteration.version,
                 IterationStatus.ANALYZED,
                 before_screenshot_key=result.get("before_screenshot_key"),
             )
@@ -287,13 +314,9 @@ async def _run_session_analysis(
             )
 
             # Auto-trigger implementation for ALL proposals
-            iteration = iter_repo.get_by_id(UUID(iteration_id))
-            if iteration:
-                iter_repo.update_status_optimistic(
-                    UUID(iteration_id),
-                    iteration.version,
-                    IterationStatus.IMPLEMENTING,
-                )
+            _update_iteration_status_with_retry(
+                iter_repo, UUID(iteration_id), IterationStatus.IMPLEMENTING
+            )
             for proposal in proposals:
                 asyncio.create_task(
                     _run_session_implementation(
@@ -310,14 +333,12 @@ async def _run_session_analysis(
                 )
         else:
             error = result.get("error", "No proposals generated")
-            iteration = iter_repo.get_by_id(UUID(iteration_id))
-            if iteration:
-                iter_repo.update_status_optimistic(
-                    UUID(iteration_id),
-                    iteration.version,
-                    IterationStatus.FAILED,
-                    error_message=error,
-                )
+            _update_iteration_status_with_retry(
+                iter_repo,
+                UUID(iteration_id),
+                IterationStatus.FAILED,
+                error_message=error,
+            )
             logger.warning(
                 "Session %s iter %d analysis failed: %s", session_id, iteration_index, error
             )
@@ -325,14 +346,12 @@ async def _run_session_analysis(
     except Exception:
         logger.exception("Session %s iter %d analysis failed", session_id, iteration_index)
         try:
-            iteration = iter_repo.get_by_id(UUID(iteration_id))
-            if iteration:
-                iter_repo.update_status_optimistic(
-                    UUID(iteration_id),
-                    iteration.version,
-                    IterationStatus.FAILED,
-                    error_message="Internal error during analysis",
-                )
+            _update_iteration_status_with_retry(
+                iter_repo,
+                UUID(iteration_id),
+                IterationStatus.FAILED,
+                error_message="Internal error during analysis",
+            )
         except Exception:
             logger.exception("Failed to update iteration status")
     finally:
@@ -576,7 +595,6 @@ def _check_iteration_completion(db: DbSession, iteration_id: UUID) -> None:
     if all_done:
         any_succeeded = any(p.status == ProposalStatus.COMPLETED for p in proposals)
         new_status = IterationStatus.COMPLETED if any_succeeded else IterationStatus.FAILED
-        iteration = iter_repo.get_by_id(iteration_id)
-        if iteration:
-            iter_repo.update_status_optimistic(iteration_id, iteration.version, new_status)
+        result = _update_iteration_status_with_retry(iter_repo, iteration_id, new_status)
+        if result:
             logger.info("Iteration %s completed: %s", iteration_id, new_status)
diff --git a/backend/app/workflow/session_analyzer_graph.py b/backend/app/workflow/session_analyzer_graph.py
index 966e210..a853396 100644
--- a/backend/app/workflow/session_analyzer_graph.py
+++ b/backend/app/workflow/session_analyzer_graph.py
@@ -22,6 +22,15 @@ async def create_k8s_job(state: SessionAnalyzerState) -> dict:
         num_proposals=state["num_proposals"],
         selected_proposal_index=state.get("selected_proposal_index"),
     )
+
+    from app.di.container import DIContainer
+
+    DIContainer.get_log_stream_service().register_job(
+        session_id=state["session_id"],
+        job_name=job_name,
+        job_type="analyze",
+    )
+
     return {"k8s_job_name": job_name, "status": "running"}
 
 
diff --git a/backend/app/workflow/session_create_pr_graph.py b/backend/app/workflow/session_create_pr_graph.py
index 9be0efb..5a705ac 100644
--- a/backend/app/workflow/session_create_pr_graph.py
+++ b/backend/app/workflow/session_create_pr_graph.py
@@ -20,6 +20,16 @@ async def create_k8s_job(state: SessionCreatePRState) -> dict:
         branch=state["branch"],
         proposal_index=state["proposal_index"],
     )
+
+    from app.di.container import DIContainer
+
+    DIContainer.get_log_stream_service().register_job(
+        session_id=state["session_id"],
+        job_name=job_name,
+        job_type="createpr",
+        proposal_index=state["proposal_index"],
+    )
+
     return {"k8s_job_name": job_name, "status": "running"}
 
 
diff --git a/backend/app/workflow/session_implementation_graph.py b/backend/app/workflow/session_implementation_graph.py
index 737d8c7..3d08d3e 100644
--- a/backend/app/workflow/session_implementation_graph.py
+++ b/backend/app/workflow/session_implementation_graph.py
@@ -28,6 +28,16 @@ async def create_k8s_job(state: SessionImplementationState) -> dict:
         proposal_plan=state["proposal_plan"],
         selected_proposal_index=state.get("selected_proposal_index"),
     )
+
+    from app.di.container import DIContainer
+
+    DIContainer.get_log_stream_service().register_job(
+        session_id=state["session_id"],
+        job_name=job_name,
+        job_type="implement",
+        proposal_index=state["proposal_index"],
+    )
+
     return {"k8s_job_name": job_name, "status": "running"}
 
 
diff --git a/backend/migration/versions/c3d4e5f6g7h8_add_session_iteration_tables.py b/backend/migration/versions/c3d4e5f6g7h8_add_session_iteration_tables.py
index 0d3d087..8a854a1 100644
--- a/backend/migration/versions/c3d4e5f6g7h8_add_session_iteration_tables.py
+++ b/backend/migration/versions/c3d4e5f6g7h8_add_session_iteration_tables.py
@@ -9,6 +9,7 @@
 
 from alembic import op
 import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
 
 
 # revision identifiers, used by Alembic.
@@ -17,6 +18,14 @@
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
+# Define enum types (create_type=False prevents auto-creation within create_table)
+sessionstatus = postgresql.ENUM("active", "completed", "archived", name="sessionstatus", create_type=False)
+iterationstatus = postgresql.ENUM(
+    "pending", "analyzing", "analyzed", "implementing", "completed", "failed",
+    name="iterationstatus", create_type=False,
+)
+proposalstatus = postgresql.ENUM("pending", "implementing", "completed", "failed", name="proposalstatus", create_type=False)
+
 
 def upgrade() -> None:
     """Drop legacy tables and create sessions, iterations, proposals tables."""
@@ -26,18 +35,18 @@ def upgrade() -> None:
     op.execute("DROP TYPE IF EXISTS proposalstatus")
     op.execute("DROP TYPE IF EXISTS jobstatus")
 
+    # Create enum types explicitly
+    sessionstatus.create(op.get_bind(), checkfirst=True)
+    iterationstatus.create(op.get_bind(), checkfirst=True)
+    proposalstatus.create(op.get_bind(), checkfirst=True)
+
     # Sessions table
     op.create_table(
         "sessions",
         sa.Column("id", sa.Uuid(), nullable=False),
         sa.Column("repo_url", sa.String(length=500), nullable=False),
         sa.Column("base_branch", sa.String(length=200), nullable=False, server_default="main"),
-        sa.Column(
-            "status",
-            sa.Enum("active", "completed", "archived", name="sessionstatus"),
-            nullable=False,
-            server_default="active",
-        ),
+        sa.Column("status", sessionstatus, nullable=False, server_default="active"),
         sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
         sa.PrimaryKeyConstraint("id"),
@@ -51,20 +60,7 @@ def upgrade() -> None:
         sa.Column("iteration_index", sa.Integer(), nullable=False),
         sa.Column("instruction", sa.Text(), nullable=False),
         sa.Column("selected_proposal_index", sa.Integer(), nullable=True),
-        sa.Column(
-            "status",
-            sa.Enum(
-                "pending",
-                "analyzing",
-                "analyzed",
-                "implementing",
-                "completed",
-                "failed",
-                name="iterationstatus",
-            ),
-            nullable=False,
-            server_default="pending",
-        ),
+        sa.Column("status", iterationstatus, nullable=False, server_default="pending"),
         sa.Column("before_screenshot_key", sa.String(length=500), nullable=True),
         sa.Column("error_message", sa.Text(), nullable=True),
         sa.Column("k8s_analyzer_job_name", sa.String(length=200), nullable=True),
@@ -91,18 +87,7 @@ def upgrade() -> None:
         sa.Column("plan", sa.Text(), nullable=False),
         sa.Column("files", sa.Text(), nullable=True),
         sa.Column("complexity", sa.String(length=20), nullable=True),
-        sa.Column(
-            "status",
-            sa.Enum(
-                "pending",
-                "implementing",
-                "completed",
-                "failed",
-                name="proposalstatus",
-            ),
-            nullable=False,
-            server_default="pending",
-        ),
+        sa.Column("status", proposalstatus, nullable=False, server_default="pending"),
         sa.Column("after_screenshot_key", sa.String(length=500), nullable=True),
         sa.Column("diff_key", sa.Text(), nullable=True),
         sa.Column("pr_url", sa.String(length=500), nullable=True),
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index b62a424..cf60855 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -13,6 +13,7 @@ dependencies = [
     "langgraph>=0.4.0",
     "kubernetes>=32.0.0",
     "boto3>=1.35.0",
+    "sse-starlette>=2.0.0",
 ]
 
 
@@ -67,7 +68,7 @@ module = "app.workflow.*"
 disallow_untyped_calls = false
 
 [[tool.mypy.overrides]]
-module = ["langgraph.*", "kubernetes.*", "pytest.*", "pytest_asyncio.*", "_pytest.*", "boto3.*", "botocore.*"]
+module = ["langgraph.*", "kubernetes.*", "pytest.*", "pytest_asyncio.*", "_pytest.*", "boto3.*", "botocore.*", "sse_starlette.*"]
 ignore_missing_imports = true
 
 [tool.hatch.build.targets.wheel]
diff --git a/backend/uv.lock b/backend/uv.lock
index 876a9f5..b51945c 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -59,6 +59,7 @@ dependencies = [
     { name = "psycopg2-binary" },
     { name = "pydantic-settings" },
     { name = "sqlalchemy" },
+    { name = "sse-starlette" },
     { name = "uvicorn" },
 ]
 
@@ -88,6 +89,7 @@ requires-dist = [
     { name = "pytest-asyncio", marker = "extra == 'test'", specifier = ">=0.24.0" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.11.0" },
     { name = "sqlalchemy", specifier = ">=2.0.46,<3.0.0" },
+    { name = "sse-starlette", specifier = ">=2.0.0" },
     { name = "uvicorn", specifier = ">=0.40.0,<1.0.0" },
 ]
 provides-extras = ["test", "dev"]
@@ -1025,6 +1027,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/fc/a1/9c4efa03300926601c19c18582531b45aededfb961ab3c3585f1e24f120b/sqlalchemy-2.0.46-py3-none-any.whl", hash = "sha256:f9c11766e7e7c0a2767dda5acb006a118640c9fc0a4104214b96269bfb78399e", size = 1937882, upload-time = "2026-01-21T18:22:10.456Z" },
 ]
 
+[[package]]
+name = "sse-starlette"
+version = "3.3.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "starlette" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5a/9f/c3695c2d2d4ef70072c3a06992850498b01c6bc9be531950813716b426fa/sse_starlette-3.3.2.tar.gz", hash = "sha256:678fca55a1945c734d8472a6cad186a55ab02840b4f6786f5ee8770970579dcd", size = 32326, upload-time = "2026-02-28T11:24:34.36Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/61/28/8cb142d3fe80c4a2d8af54ca0b003f47ce0ba920974e7990fa6e016402d1/sse_starlette-3.3.2-py3-none-any.whl", hash = "sha256:5c3ea3dad425c601236726af2f27689b74494643f57017cafcb6f8c9acfbb862", size = 14270, upload-time = "2026-02-28T11:24:32.984Z" },
+]
+
 [[package]]
 name = "starlette"
 version = "0.52.1"
diff --git a/docker/worker-analyze.py b/docker/worker-analyze.py
index 3ab462d..00fbd59 100644
--- a/docker/worker-analyze.py
+++ b/docker/worker-analyze.py
@@ -10,11 +10,45 @@
 import re
 import subprocess
 import sys
+import time
 from pathlib import Path
 
 import boto3
 from botocore.config import Config
-from claude_agent_sdk import ClaudeAgentOptions, query
+from claude_agent_sdk import ClaudeAgentOptions, query, AssistantMessage, TextBlock, ToolUseBlock
+from claude_agent_sdk.types import StreamEvent
+
+
+def emit_log(phase: str, message: str, detail: str | None = None) -> None:
+    entry = {
+        "phase": phase,
+        "message": message,
+        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    }
+    if detail:
+        entry["detail"] = detail
+    print(f"@@LOG@@{json.dumps(entry, ensure_ascii=False)}", flush=True)
+
+
+def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
+    """Parse tool input JSON and emit a human-readable log line."""
+    try:
+        params = json.loads(raw_input)
+    except json.JSONDecodeError:
+        emit_log(phase, f"Using tool: {tool_name}")
+        return
+
+    if tool_name == "Read":
+        emit_log(phase, f"Reading: {params.get('file_path', '?')}")
+    elif tool_name in ("Write", "Edit"):
+        emit_log(phase, f"Editing: {params.get('file_path', '?')}")
+    elif tool_name == "Bash":
+        cmd = params.get("command", "?")
+        emit_log(phase, f"Running: {cmd[:100]}")
+    elif tool_name in ("Glob", "Grep"):
+        emit_log(phase, f"Searching: {params.get('pattern', '?')}")
+    else:
+        emit_log(phase, f"Using tool: {tool_name}")
 
 
 def get_s3_client():
@@ -87,6 +121,9 @@ async def take_before_screenshot(repo_dir: str, screenshot_output: str) -> None:
 IMPORTANT: The screenshot is critical. Do your best to get the dev server running and take the screenshot.
 """
 
+    current_tool = None
+    tool_input_chunks = ""
+
     async for msg in query(
         prompt=prompt,
         options=ClaudeAgentOptions(
@@ -94,12 +131,37 @@ async def take_before_screenshot(repo_dir: str, screenshot_output: str) -> None:
             cwd=repo_dir,
             max_turns=20,
             max_budget_usd=1.0,
+            include_partial_messages=True,
         ),
     ):
-        if hasattr(msg, "content"):
+        if isinstance(msg, StreamEvent):
+            event = msg.event
+            etype = event.get("type")
+            if etype == "content_block_start":
+                content_block = event.get("content_block", {})
+                if content_block.get("type") == "tool_use":
+                    current_tool = content_block.get("name")
+                    tool_input_chunks = ""
+                    emit_log("screenshot", f"Tool: {current_tool}")
+            elif etype == "content_block_delta":
+                delta = event.get("delta", {})
+                if delta.get("type") == "input_json_delta":
+                    tool_input_chunks += delta.get("partial_json", "")
+            elif etype == "content_block_stop":
+                if current_tool and tool_input_chunks:
+                    _emit_tool_detail("screenshot", current_tool, tool_input_chunks)
+                current_tool = None
+                tool_input_chunks = ""
+            continue
+
+        if isinstance(msg, AssistantMessage):
             for block in msg.content:
-                if hasattr(block, "text"):
+                if isinstance(block, TextBlock):
+                    emit_log("screenshot", block.text[:200], detail=block.text)
                     print(f"Screenshot Agent: {block.text[:200]}")
+                elif isinstance(block, ToolUseBlock):
+                    _emit_tool_detail("screenshot", block.name, json.dumps(block.input))
+                    print(f"Screenshot Tool: {block.name}")
 
 
 def _extract_proposals_json(collected_text: list[str]) -> list[dict]:
@@ -115,6 +177,8 @@ def _extract_proposals_json(collected_text: list[str]) -> list[dict]:
             text = text.split("```json")[1].split("```")[0].strip()
         elif "```" in text:
             text = text.split("```")[1].split("```")[0].strip()
+
+        # Try parsing the whole block first
         try:
             data = json.loads(text)
             if isinstance(data, dict) and "proposals" in data:
@@ -122,15 +186,29 @@ def _extract_proposals_json(collected_text: list[str]) -> list[dict]:
             if isinstance(data, list):
                 return data
         except json.JSONDecodeError:
-            continue
+            pass
+
+        # If block contains JSON embedded after prose, find the first '{' and try parsing from there
+        brace_idx = text.find('{"proposals"')
+        if brace_idx == -1:
+            brace_idx = text.find("{\n")
+        if brace_idx >= 0:
+            candidate = text[brace_idx:]
+            try:
+                data = json.loads(candidate)
+                if isinstance(data, dict) and "proposals" in data:
+                    return data["proposals"]
+            except json.JSONDecodeError:
+                pass
 
-    # Strategy 2: Search for {"proposals": ...} in the concatenated text using regex
+    # Strategy 2: Find {"proposals" in concatenated text and use JSONDecoder to parse
     full_text = "\n".join(collected_text)
-    match = re.search(r'\{"proposals"\s*:\s*\[.*\]\s*\}', full_text, re.DOTALL)
-    if match:
+    decoder = json.JSONDecoder()
+    idx = full_text.find('{"proposals"')
+    if idx >= 0:
         try:
-            data = json.loads(match.group())
-            if "proposals" in data:
+            data, _ = decoder.raw_decode(full_text, idx)
+            if isinstance(data, dict) and "proposals" in data:
                 return data["proposals"]
         except json.JSONDecodeError:
             pass
@@ -193,19 +271,47 @@ async def generate_proposals(
 """
 
     collected_text = []
+    current_tool = None
+    tool_input_chunks = ""
+
     async for msg in query(
         prompt=prompt,
         options=ClaudeAgentOptions(
             allowed_tools=["Read", "Glob", "Grep"],
             cwd=repo_dir,
             max_turns=30,
+            include_partial_messages=True,
         ),
     ):
+        if isinstance(msg, StreamEvent):
+            event = msg.event
+            etype = event.get("type")
+            if etype == "content_block_start":
+                content_block = event.get("content_block", {})
+                if content_block.get("type") == "tool_use":
+                    current_tool = content_block.get("name")
+                    tool_input_chunks = ""
+                    emit_log("analyzing", f"Tool: {current_tool}")
+            elif etype == "content_block_delta":
+                delta = event.get("delta", {})
+                if delta.get("type") == "input_json_delta":
+                    tool_input_chunks += delta.get("partial_json", "")
+            elif etype == "content_block_stop":
+                if current_tool and tool_input_chunks:
+                    _emit_tool_detail("analyzing", current_tool, tool_input_chunks)
+                current_tool = None
+                tool_input_chunks = ""
+            continue
+
         # Collect text content from assistant messages
-        if hasattr(msg, "content"):
+        if isinstance(msg, AssistantMessage):
             for block in msg.content:
-                if hasattr(block, "text"):
+                if isinstance(block, TextBlock):
+                    emit_log("analyzing", block.text[:200], detail=block.text)
                     collected_text.append(block.text)
+                elif isinstance(block, ToolUseBlock):
+                    _emit_tool_detail("analyzing", block.name, json.dumps(block.input))
+                    print(f"Analyze Tool: {block.name}")
 
     return _extract_proposals_json(collected_text)
 
@@ -229,7 +335,7 @@ async def main() -> None:
     s3_prefix = f"sessions/{session_id}/iterations/{iteration_index}"
 
     # Step 1: Clone repository
-    print("=== Cloning repository ===")
+    emit_log("cloning", "Cloning repository")
     subprocess.run(
         ["git", "clone", "--depth", "1", "--branch", branch, repo_url, repo_dir],
         check=True,
@@ -243,9 +349,9 @@ async def main() -> None:
             f"/proposals/{selected_proposal_index}/changes.diff"
         )
         local_patch = f"{tmp_dir}/parent.diff"
-        print(f"=== Downloading patch from S3: {patch_key} ===")
+        emit_log("patching", f"Downloading patch from S3: {patch_key}")
         if s3_download(s3, bucket, patch_key, local_patch):
-            print("=== Applying cumulative patch ===")
+            emit_log("patching", "Applying cumulative patch")
             result = subprocess.run(
                 ["git", "am", "--3way", local_patch],
                 cwd=repo_dir,
@@ -269,7 +375,7 @@ async def main() -> None:
                     cwd=repo_dir,
                     check=True,
                 )
-            print("=== Cumulative patch applied successfully ===")
+            emit_log("patching", "Cumulative patch applied successfully")
         else:
             print(
                 f"WARNING: Patch not found at s3://{bucket}/{patch_key}",
@@ -277,17 +383,17 @@ async def main() -> None:
             )
 
     # Step 2: Take before screenshot
-    print("=== Taking before screenshot ===")
+    emit_log("screenshot", "Taking before screenshot")
     before_path = f"{tmp_dir}/before.png"
     await take_before_screenshot(repo_dir, before_path)
 
     # Step 3: Generate design proposals via Claude Agent SDK
-    print("=== Generating design proposals ===")
+    emit_log("analyzing", "Generating design proposals")
     proposals = await generate_proposals(repo_dir, instruction, num_proposals)
-    print(f"=== Generated {len(proposals)} proposals ===")
+    emit_log("analyzing", f"Generated {len(proposals)} proposals")
 
     # Step 4: Upload results to S3
-    print("=== Uploading results to S3 ===")
+    emit_log("uploading", "Uploading results to S3")
 
     # Upload before screenshot
     if Path(before_path).exists():
@@ -307,7 +413,7 @@ async def main() -> None:
         content_type="application/json",
     )
 
-    print("=== Analysis Complete ===")
+    emit_log("completed", "Analysis complete")
 
 
 if __name__ == "__main__":
diff --git a/docker/worker-createpr.py b/docker/worker-createpr.py
index 20d9071..b428d8e 100644
--- a/docker/worker-createpr.py
+++ b/docker/worker-createpr.py
@@ -5,16 +5,51 @@
 """
 
 import asyncio
+import json
 import os
 import re
 import subprocess
 import sys
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 
 import boto3
 from botocore.config import Config
-from claude_agent_sdk import ClaudeAgentOptions, query
+from claude_agent_sdk import ClaudeAgentOptions, query, AssistantMessage, TextBlock, ToolUseBlock
+from claude_agent_sdk.types import StreamEvent
+
+
+def emit_log(phase: str, message: str, detail: str | None = None) -> None:
+    entry = {
+        "phase": phase,
+        "message": message,
+        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    }
+    if detail:
+        entry["detail"] = detail
+    print(f"@@LOG@@{json.dumps(entry, ensure_ascii=False)}", flush=True)
+
+
+def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
+    """Parse tool input JSON and emit a human-readable log line."""
+    try:
+        params = json.loads(raw_input)
+    except json.JSONDecodeError:
+        emit_log(phase, f"Using tool: {tool_name}")
+        return
+
+    if tool_name == "Read":
+        emit_log(phase, f"Reading: {params.get('file_path', '?')}")
+    elif tool_name in ("Write", "Edit"):
+        emit_log(phase, f"Editing: {params.get('file_path', '?')}")
+    elif tool_name == "Bash":
+        cmd = params.get("command", "?")
+        emit_log(phase, f"Running: {cmd[:100]}")
+    elif tool_name in ("Glob", "Grep"):
+        emit_log(phase, f"Searching: {params.get('pattern', '?')}")
+    else:
+        emit_log(phase, f"Using tool: {tool_name}")
 
 
 def get_s3_client():
@@ -87,18 +122,46 @@ async def push_and_create_pr(
 IMPORTANT: You MUST output the PR URL in that exact format so it can be extracted.
 """
     collected_text: list[str] = []
+    current_tool = None
+    tool_input_chunks = ""
+
     async for msg in query(
         prompt=prompt,
         options=ClaudeAgentOptions(
             allowed_tools=["Read", "Bash", "Glob", "Grep"],
             cwd=repo_dir, max_turns=15, max_budget_usd=1.0,
+            include_partial_messages=True,
         ),
     ):
-        if hasattr(msg, "content"):
+        if isinstance(msg, StreamEvent):
+            event = msg.event
+            etype = event.get("type")
+            if etype == "content_block_start":
+                content_block = event.get("content_block", {})
+                if content_block.get("type") == "tool_use":
+                    current_tool = content_block.get("name")
+                    tool_input_chunks = ""
+                    emit_log("creating_pr", f"Tool: {current_tool}")
+            elif etype == "content_block_delta":
+                delta = event.get("delta", {})
+                if delta.get("type") == "input_json_delta":
+                    tool_input_chunks += delta.get("partial_json", "")
+            elif etype == "content_block_stop":
+                if current_tool and tool_input_chunks:
+                    _emit_tool_detail("creating_pr", current_tool, tool_input_chunks)
+                current_tool = None
+                tool_input_chunks = ""
+            continue
+
+        if isinstance(msg, AssistantMessage):
             for block in msg.content:
-                if hasattr(block, "text"):
+                if isinstance(block, TextBlock):
                     collected_text.append(block.text)
+                    emit_log("creating_pr", block.text[:200], detail=block.text)
                     print(f"Agent: {block.text[:200]}")
+                elif isinstance(block, ToolUseBlock):
+                    _emit_tool_detail("creating_pr", block.name, json.dumps(block.input))
+                    print(f"Agent Tool: {block.name}")
 
     full_text = "\n".join(collected_text)
     for line in full_text.split("\n"):
@@ -136,10 +199,10 @@ async def main() -> None:
         print(f"Error: Patch not found at s3://{bucket}/{patch_key}", file=sys.stderr)
         sys.exit(1)
     diff_content = Path(local_patch).read_text()
-    print(f"=== Patch loaded ({len(diff_content)} chars) ===")
+    emit_log("cloning", f"Patch loaded ({len(diff_content)} chars)")
 
     # Step 2: Clone repository (with token auth for push)
-    print("=== Cloning repository ===")
+    emit_log("cloning", "Cloning repository")
     repo_dir = "/workspace/repo"
     if repo_url.startswith("https://github.com/"):
         auth_repo_url = repo_url.replace(
@@ -157,11 +220,11 @@ async def main() -> None:
     # Step 3: Create a new branch
     ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
     branch_name = f"feat/{ts}-ui-session-{session_id[:8]}"
-    print(f"=== Creating branch: {branch_name} ===")
+    emit_log("patching", f"Creating branch: {branch_name}")
     subprocess.run(["git", "checkout", "-b", branch_name], cwd=repo_dir, check=True)
 
     # Step 4: Apply the patch
-    print("=== Applying patch ===")
+    emit_log("patching", "Applying patch")
     result = subprocess.run(
         ["git", "am", "--3way", local_patch],
         cwd=repo_dir, capture_output=True, text=True,
@@ -181,7 +244,7 @@ async def main() -> None:
     subprocess.run(["git", "fetch", "--unshallow"], cwd=repo_dir, capture_output=True)
 
     # Step 6: Push and create PR via Agent SDK
-    print("=== Creating PR via Agent SDK ===")
+    emit_log("pushing", "Pushing branch and creating PR")
     diff_summary = diff_content[:10000]
     pr_url = await push_and_create_pr(repo_dir, branch_name, branch, diff_summary)
 
@@ -196,7 +259,7 @@ async def main() -> None:
         print("Error: Failed to extract PR URL from agent output", file=sys.stderr)
         sys.exit(1)
 
-    print("=== PR Creation Complete ===")
+    emit_log("completed", "PR creation complete")
 
 
 if __name__ == "__main__":
diff --git a/docker/worker-implement.py b/docker/worker-implement.py
index d43ab3f..17779cf 100644
--- a/docker/worker-implement.py
+++ b/docker/worker-implement.py
@@ -10,11 +10,45 @@
 import os
 import subprocess
 import sys
+import time
 from pathlib import Path
 
 import boto3
 from botocore.config import Config
-from claude_agent_sdk import ClaudeAgentOptions, query
+from claude_agent_sdk import ClaudeAgentOptions, query, AssistantMessage, TextBlock, ToolUseBlock
+from claude_agent_sdk.types import StreamEvent
+
+
+def emit_log(phase: str, message: str, detail: str | None = None) -> None:
+    entry = {
+        "phase": phase,
+        "message": message,
+        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+    }
+    if detail:
+        entry["detail"] = detail
+    print(f"@@LOG@@{json.dumps(entry, ensure_ascii=False)}", flush=True)
+
+
+def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
+    """Parse tool input JSON and emit a human-readable log line."""
+    try:
+        params = json.loads(raw_input)
+    except json.JSONDecodeError:
+        emit_log(phase, f"Using tool: {tool_name}")
+        return
+
+    if tool_name == "Read":
+        emit_log(phase, f"Reading: {params.get('file_path', '?')}")
+    elif tool_name in ("Write", "Edit"):
+        emit_log(phase, f"Editing: {params.get('file_path', '?')}")
+    elif tool_name == "Bash":
+        cmd = params.get("command", "?")
+        emit_log(phase, f"Running: {cmd[:100]}")
+    elif tool_name in ("Glob", "Grep"):
+        emit_log(phase, f"Searching: {params.get('pattern', '?')}")
+    else:
+        emit_log(phase, f"Using tool: {tool_name}")
 
 
 def get_s3_client():
@@ -102,6 +136,9 @@ async def implement_proposal(
 IMPORTANT: The screenshot is critical. Do your best to get the dev server running and take the screenshot.
 """
 
+    current_tool = None
+    tool_input_chunks = ""
+
     async for msg in query(
         prompt=prompt,
         options=ClaudeAgentOptions(
@@ -109,13 +146,38 @@ async def implement_proposal(
             cwd=repo_dir,
             max_turns=40,
             max_budget_usd=3.0,
+            include_partial_messages=True,
         ),
     ):
-        # Log progress messages
-        if hasattr(msg, "content"):
+        if isinstance(msg, StreamEvent):
+            event = msg.event
+            etype = event.get("type")
+            if etype == "content_block_start":
+                content_block = event.get("content_block", {})
+                if content_block.get("type") == "tool_use":
+                    current_tool = content_block.get("name")
+                    tool_input_chunks = ""
+                    emit_log("implementing", f"Tool: {current_tool}")
+            elif etype == "content_block_delta":
+                delta = event.get("delta", {})
+                if delta.get("type") == "input_json_delta":
+                    tool_input_chunks += delta.get("partial_json", "")
+            elif etype == "content_block_stop":
+                if current_tool and tool_input_chunks:
+                    _emit_tool_detail("implementing", current_tool, tool_input_chunks)
+                current_tool = None
+                tool_input_chunks = ""
+            continue
+
+        # Log progress messages from completed AssistantMessage
+        if isinstance(msg, AssistantMessage):
             for block in msg.content:
-                if hasattr(block, "text"):
+                if isinstance(block, TextBlock):
+                    emit_log("implementing", block.text[:200], detail=block.text)
                     print(f"Agent: {block.text[:200]}")
+                elif isinstance(block, ToolUseBlock):
+                    _emit_tool_detail("implementing", block.name, json.dumps(block.input))
+                    print(f"Agent Tool: {block.name}")
 
 
 async def main() -> None:
@@ -141,7 +203,7 @@ async def main() -> None:
     # Step 1: Download proposal plan from S3
     plan_key = f"{s3_prefix}/plan.json"
     local_plan = f"{tmp_dir}/plan.json"
-    print(f"=== Downloading plan from S3: {plan_key} ===")
+    emit_log("cloning", f"Downloading plan from S3: {plan_key}")
     if s3_download(s3, bucket, plan_key, local_plan):
         proposal_plan = Path(local_plan).read_text()
     else:
@@ -153,7 +215,7 @@ async def main() -> None:
         sys.exit(1)
 
     # Step 2: Clone repository
-    print("=== Cloning repository ===")
+    emit_log("cloning", "Cloning repository")
     subprocess.run(
         ["git", "clone", "--depth", "1", "--branch", branch, repo_url, repo_dir],
         check=True,
@@ -167,9 +229,9 @@ async def main() -> None:
             f"/proposals/{selected_proposal_index}/changes.diff"
         )
         local_patch = f"{tmp_dir}/parent.diff"
-        print(f"=== Downloading patch from S3: {patch_key} ===")
+        emit_log("patching", f"Downloading patch from S3: {patch_key}")
         if s3_download(s3, bucket, patch_key, local_patch):
-            print("=== Applying cumulative patch ===")
+            emit_log("patching", "Applying cumulative patch")
             result = subprocess.run(
                 ["git", "am", "--3way", local_patch],
                 cwd=repo_dir,
@@ -193,7 +255,7 @@ async def main() -> None:
                     cwd=repo_dir,
                     check=True,
                 )
-            print("=== Cumulative patch applied successfully ===")
+            emit_log("patching", "Cumulative patch applied successfully")
         else:
             print(
                 f"WARNING: Patch not found at s3://{bucket}/{patch_key}",
@@ -203,14 +265,14 @@ async def main() -> None:
     # Step 3: Create local branch and implement the proposal
     branch_name = f"local/session-{session_id[:8]}-iter{iteration_index}-prop{proposal_index}"
     subprocess.run(["git", "checkout", "-b", branch_name], cwd=repo_dir, check=True)
-    print(f"=== Created local branch: {branch_name} ===")
+    emit_log("implementing", f"Created local branch: {branch_name}")
 
     screenshot_path = f"{tmp_dir}/after.png"
-    print("=== Implementing design proposal ===")
+    emit_log("implementing", "Implementing design proposal")
     await implement_proposal(repo_dir, proposal_plan, screenshot_path)
 
     # Step 4: Generate cumulative patch (squash everything since base_branch)
-    print("=== Generating cumulative patch ===")
+    emit_log("uploading", "Generating cumulative patch")
     # Parse proposal plan to get title for commit message
     try:
         plan_data = json.loads(proposal_plan)
@@ -242,7 +304,7 @@ async def main() -> None:
         f.write(patch_result.stdout)
 
     # Step 5: Upload results to S3
-    print("=== Uploading results to S3 ===")
+    emit_log("uploading", "Uploading results to S3")
 
     # Upload after screenshot
     if Path(screenshot_path).exists():
@@ -261,7 +323,7 @@ async def main() -> None:
         content_type="text/plain",
     )
 
-    print("=== Implementation Complete ===")
+    emit_log("completed", "Implementation complete")
 
 
 if __name__ == "__main__":
diff --git a/frontend/src/components/LogPanel.tsx b/frontend/src/components/LogPanel.tsx
new file mode 100644
index 0000000..bd728b6
--- /dev/null
+++ b/frontend/src/components/LogPanel.tsx
@@ -0,0 +1,275 @@
+import { useState, useRef, useEffect, useCallback } from 'react'
+import type { LogStreamState, JobLogState, LogEntry } from '../hooks/useLogStream'
+
+const PHASE_LABELS: Record<string, string> = {
+  cloning: 'Cloning',
+  patching: 'Applying Patch',
+  screenshot: 'Taking Screenshot',
+  analyzing: 'Analyzing',
+  implementing: 'Implementing',
+  uploading: 'Uploading',
+  pushing: 'Pushing',
+  creating_pr: 'Creating PR',
+  completed: 'Completed',
+  waiting: 'Waiting',
+  running: 'Running',
+  error: 'Error',
+}
+
+const JOB_TYPE_LABELS: Record<string, string> = {
+  analyze: 'Analyze',
+  implement: 'Implement',
+  createpr: 'Create PR',
+}
+
+function getPhaseLabel(phase: string): string {
+  return PHASE_LABELS[phase] || phase
+}
+
+function getJobLabel(jobKey: string): string {
+  const [jobType, indexStr] = jobKey.split(':')
+  const base = JOB_TYPE_LABELS[jobType] || jobType
+  if (indexStr != null) {
+    return `${base} #${Number(indexStr) + 1}`
+  }
+  return base
+}
+
+function getOverallPhase(logState: LogStreamState): string {
+  // Find the most recent active phase across all jobs
+  let latestPhase = 'waiting'
+  for (const [, job] of logState.jobs) {
+    if (job.phase === 'completed') continue
+    if (job.phase === 'error') return 'error'
+    latestPhase = job.phase
+  }
+  // If all are completed
+  const allCompleted = logState.jobs.size > 0 && [...logState.jobs.values()].every((j) => j.phase === 'completed')
+  if (allCompleted) return 'completed'
+  return latestPhase
+}
+
+const ROW_HEIGHT = 20
+
+interface LogViewerProps {
+  entries: LogEntry[]
+}
+
+function LogViewer({ entries }: LogViewerProps) {
+  const containerRef = useRef<HTMLDivElement>(null)
+  const [scrollTop, setScrollTop] = useState(0)
+  const [autoScroll, setAutoScroll] = useState(true)
+
+  const totalHeight = entries.length * ROW_HEIGHT
+  const containerHeight = 300
+
+  // Auto-scroll to bottom when new entries arrive
+  useEffect(() => {
+    if (autoScroll && containerRef.current) {
+      containerRef.current.scrollTop = totalHeight - containerHeight
+    }
+  }, [entries.length, autoScroll, totalHeight])
+
+  const handleScroll = useCallback(() => {
+    if (!containerRef.current) return
+    const { scrollTop: st, scrollHeight, clientHeight } = containerRef.current
+    setScrollTop(st)
+    // If user scrolls near the bottom, re-enable auto-scroll
+    const isNearBottom = scrollHeight - st - clientHeight < ROW_HEIGHT * 3
+    setAutoScroll(isNearBottom)
+  }, [])
+
+  const startIndex = Math.max(0, Math.floor(scrollTop / ROW_HEIGHT) - 5)
+  const endIndex = Math.min(entries.length, Math.ceil((scrollTop + containerHeight) / ROW_HEIGHT) + 5)
+  const visibleEntries = entries.slice(startIndex, endIndex)
+
+  return (
+    <div
+      ref={containerRef}
+      onScroll={handleScroll}
+      style={{
+        height: `${containerHeight}px`,
+        overflow: 'auto',
+        position: 'relative',
+        backgroundColor: '#0d1117',
+        fontFamily: 'ui-monospace, SFMono-Regular, "SF Mono", Menlo, monospace',
+        fontSize: '12px',
+        lineHeight: `${ROW_HEIGHT}px`,
+      }}
+    >
+      <div style={{ height: `${totalHeight}px`, position: 'relative' }}>
+        {visibleEntries.map((entry, i) => {
+          const index = startIndex + i
+          const time = entry.timestamp
+            ? new Date(entry.timestamp).toLocaleTimeString()
+            : ''
+          return (
+            <div
+              key={index}
+              style={{
+                position: 'absolute',
+                top: `${index * ROW_HEIGHT}px`,
+                left: 0,
+                right: 0,
+                height: `${ROW_HEIGHT}px`,
+                padding: '0 8px',
+                whiteSpace: 'nowrap',
+                overflow: 'hidden',
+                textOverflow: 'ellipsis',
+                color: entry.phase === 'error' ? '#f87171' : '#c9d1d9',
+              }}
+            >
+              <span style={{ color: '#6e7681', marginRight: '8px' }}>{time}</span>
+              <span>{entry.message}</span>
+            </div>
+          )
+        })}
+      </div>
+    </div>
+  )
+}
+
+interface LogPanelProps {
+  logState: LogStreamState
+}
+
+export default function LogPanel({ logState }: LogPanelProps) {
+  const [expanded, setExpanded] = useState(false)
+  const [activeTab, setActiveTab] = useState<string | null>(null)
+
+  const jobKeys = [...logState.jobs.keys()]
+
+  // Auto-select first tab
+  useEffect(() => {
+    if (activeTab === null && jobKeys.length > 0) {
+      setActiveTab(jobKeys[0])
+    }
+  }, [jobKeys.length, activeTab, jobKeys])
+
+  const overallPhase = getOverallPhase(logState)
+  const phaseLabel = getPhaseLabel(overallPhase)
+
+  const activeJobState: JobLogState | undefined =
+    activeTab ? logState.jobs.get(activeTab) : undefined
+
+  const dotColor =
+    overallPhase === 'completed'
+      ? '#34d399'
+      : overallPhase === 'error'
+        ? '#f87171'
+        : '#60a5fa'
+
+  return (
+    <div
+      style={{
+        borderRadius: '8px',
+        border: '1px solid #30363d',
+        overflow: 'hidden',
+        marginBottom: '16px',
+        backgroundColor: '#161b22',
+      }}
+    >
+      {/* Status line */}
+      <button
+        onClick={() => setExpanded(!expanded)}
+        style={{
+          width: '100%',
+          display: 'flex',
+          alignItems: 'center',
+          justifyContent: 'space-between',
+          padding: '12px 16px',
+          backgroundColor: 'transparent',
+          border: 'none',
+          cursor: 'pointer',
+          color: '#c9d1d9',
+          fontSize: '14px',
+          fontWeight: 500,
+        }}
+      >
+        <span style={{ display: 'flex', alignItems: 'center', gap: '8px' }}>
+          <span
+            style={{
+              width: '8px',
+              height: '8px',
+              borderRadius: '50%',
+              backgroundColor: dotColor,
+              display: 'inline-block',
+              animation:
+                overallPhase !== 'completed' && overallPhase !== 'error'
+                  ? 'pulse 2s infinite'
+                  : 'none',
+            }}
+          />
+          {phaseLabel}
+          {logState.isStreaming && overallPhase !== 'completed' && '...'}
+        </span>
+        <span style={{ color: '#6e7681', fontSize: '12px' }}>
+          {expanded ? 'Hide' : 'Details'}
+        </span>
+      </button>
+
+      {/* Expanded content */}
+      {expanded && (
+        <div>
+          {/* Tab bar (show only if multiple jobs) */}
+          {jobKeys.length > 1 && (
+            <div
+              style={{
+                display: 'flex',
+                gap: '0',
+                borderBottom: '1px solid #30363d',
+                padding: '0 8px',
+              }}
+            >
+              {jobKeys.map((key) => (
+                <button
+                  key={key}
+                  onClick={() => setActiveTab(key)}
+                  style={{
+                    padding: '8px 16px',
+                    border: 'none',
+                    borderBottom:
+                      activeTab === key ? '2px solid #58a6ff' : '2px solid transparent',
+                    backgroundColor: 'transparent',
+                    color: activeTab === key ? '#c9d1d9' : '#6e7681',
+                    cursor: 'pointer',
+                    fontSize: '13px',
+                    fontWeight: activeTab === key ? 600 : 400,
+                  }}
+                >
+                  {getJobLabel(key)}
+                </button>
+              ))}
+            </div>
+          )}
+
+          {/* Log viewer */}
+          {activeJobState ? (
+            <LogViewer entries={activeJobState.entries} />
+          ) : (
+            <div
+              style={{
+                height: '100px',
+                display: 'flex',
+                alignItems: 'center',
+                justifyContent: 'center',
+                color: '#6e7681',
+                fontSize: '13px',
+              }}
+            >
+              Waiting for logs...
+            </div>
+          )}
+        </div>
+      )}
+
+      {/* Pulse animation */}
+      <style>{`
+        @keyframes pulse {
+          0%, 100% { opacity: 1; }
+          50% { opacity: 0.4; }
+        }
+      `}</style>
+    </div>
+  )
+}
diff --git a/frontend/src/hooks/useLogStream.ts b/frontend/src/hooks/useLogStream.ts
new file mode 100644
index 0000000..4b09a84
--- /dev/null
+++ b/frontend/src/hooks/useLogStream.ts
@@ -0,0 +1,150 @@
+import { useState, useEffect, useRef, useCallback } from 'react'
+
+export interface LogEntry {
+  job_type: string
+  proposal_index: number | null
+  phase: string
+  message: string
+  detail?: string
+  timestamp?: string
+}
+
+export interface JobLogState {
+  phase: string
+  entries: LogEntry[]
+}
+
+export interface LogStreamState {
+  /** Map of jobKey -> log state. jobKey = `${job_type}` or `${job_type}:${proposal_index}` */
+  jobs: Map<string, JobLogState>
+  isStreaming: boolean
+  isDone: boolean
+}
+
+const MAX_ENTRIES_PER_JOB = 5000
+
+function jobKey(entry: LogEntry): string {
+  if (entry.proposal_index != null) {
+    return `${entry.job_type}:${entry.proposal_index}`
+  }
+  return entry.job_type
+}
+
+export function useLogStream(sessionId: string | null, enabled: boolean): LogStreamState {
+  const [state, setState] = useState<LogStreamState>({
+    jobs: new Map(),
+    isStreaming: false,
+    isDone: false,
+  })
+  const eventSourceRef = useRef<EventSource | null>(null)
+  const jobsRef = useRef<Map<string, JobLogState>>(new Map())
+  const connectedOnceRef = useRef(false)
+  const enabledRef = useRef(enabled)
+  const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
+  const [reconnectCount, setReconnectCount] = useState(0)
+
+  // Keep enabledRef in sync
+  enabledRef.current = enabled
+
+  const cleanup = useCallback(() => {
+    if (eventSourceRef.current) {
+      eventSourceRef.current.close()
+      eventSourceRef.current = null
+    }
+    if (reconnectTimerRef.current) {
+      clearTimeout(reconnectTimerRef.current)
+      reconnectTimerRef.current = null
+    }
+  }, [])
+
+  // Reset connectedOnceRef when sessionId changes
+  useEffect(() => {
+    connectedOnceRef.current = false
+  }, [sessionId])
+
+  useEffect(() => {
+    if (!sessionId || !enabled) {
+      cleanup()
+      return
+    }
+
+    // On reconnect after done, preserve existing logs
+    const isReconnect = connectedOnceRef.current
+    if (!isReconnect) {
+      jobsRef.current = new Map()
+      setState({ jobs: new Map(), isStreaming: true, isDone: false })
+    } else {
+      setState((prev) => ({ ...prev, isStreaming: true, isDone: false }))
+    }
+
+    // On reconnect (2nd+ connection), skip old logs by requesting only recent ones
+    const url = isReconnect
+      ? `/api/sessions/${sessionId}/logs/stream?since_seconds=1`
+      : `/api/sessions/${sessionId}/logs/stream`
+    connectedOnceRef.current = true
+
+    const es = new EventSource(url)
+    eventSourceRef.current = es
+
+    es.addEventListener('log', (event: MessageEvent) => {
+      try {
+        const data = JSON.parse(event.data) as LogEntry
+        // Skip keepalive events
+        if (data.job_type === '_keepalive') return
+
+        const key = jobKey(data)
+        const current = jobsRef.current.get(key) || { phase: 'pending', entries: [] }
+
+        // Update phase
+        current.phase = data.phase
+
+        // Append entry with rotation
+        current.entries.push(data)
+        if (current.entries.length > MAX_ENTRIES_PER_JOB) {
+          current.entries = current.entries.slice(-MAX_ENTRIES_PER_JOB)
+        }
+
+        jobsRef.current.set(key, current)
+        setState({
+          jobs: new Map(jobsRef.current),
+          isStreaming: true,
+          isDone: false,
+        })
+      } catch {
+        // Ignore parse errors
+      }
+    })
+
+    es.addEventListener('done', () => {
+      if (eventSourceRef.current) {
+        eventSourceRef.current.close()
+        eventSourceRef.current = null
+      }
+      // If still enabled (iteration in progress), auto-reconnect after delay
+      if (enabledRef.current) {
+        setState((prev) => ({ ...prev, isStreaming: false }))
+        reconnectTimerRef.current = setTimeout(() => {
+          reconnectTimerRef.current = null
+          setReconnectCount((c) => c + 1)
+        }, 2000)
+      } else {
+        setState((prev) => ({ ...prev, isStreaming: false, isDone: true }))
+      }
+    })
+
+    es.onerror = () => {
+      // EventSource will auto-reconnect for network errors
+      // If readyState is CLOSED, the connection has been permanently closed
+      if (es.readyState === EventSource.CLOSED) {
+        setState((prev) => ({
+          ...prev,
+          isStreaming: false,
+        }))
+      }
+    }
+
+    return cleanup
+  }, [sessionId, enabled, cleanup, reconnectCount])
+
+  return state
+}
diff --git a/frontend/src/pages/SessionDetail.tsx b/frontend/src/pages/SessionDetail.tsx
index 7c4b9dc..fa8901f 100644
--- a/frontend/src/pages/SessionDetail.tsx
+++ b/frontend/src/pages/SessionDetail.tsx
@@ -1,16 +1,28 @@
 import { useState, useCallback, useEffect, useRef } from 'react'
 import { useParams } from 'react-router-dom'
 import { useSessionPolling } from '../hooks/useSessionPolling'
+import { useLogStream } from '../hooks/useLogStream'
 import { createPR, iterate, getSession } from '../services/api'
 import type { Iteration, Proposal } from '../services/api'
 import { useLayoutContext } from '../hooks/useLayoutContext'
 import StatusBadge from '../components/StatusBadge'
 import ProposalCard from '../components/ProposalCard'
+import LogPanel from '../components/LogPanel'
 
 export default function SessionDetail() {
   const { sessionId } = useParams<{ sessionId: string }>()
   const { refreshSessions } = useLayoutContext()
   const { session, error, isLoading, refetch } = useSessionPolling(sessionId ?? null)
+
+  // Compute isInProgress early so useLogStream can be called unconditionally
+  const latestIterationForHook =
+    session && session.iterations.length > 0
+      ? session.iterations[session.iterations.length - 1]
+      : null
+  const iterationStatusForHook = latestIterationForHook?.status ?? 'pending'
+  const isInProgressForHook = ['pending', 'analyzing', 'implementing'].includes(iterationStatusForHook)
+  const logStreamState = useLogStream(sessionId ?? null, isInProgressForHook)
+
   const [selectedIndex, setSelectedIndex] = useState<number | null>(null)
   const [prLoading, setPrLoading] = useState(false)
   const [prError, setPrError] = useState<string | null>(null)
@@ -222,25 +234,8 @@ export default function SessionDetail() {
         </div>
       )}
 
-      {/* Progress indicator */}
-      {isInProgress && (
-        <div
-          style={{
-            padding: '16px',
-            backgroundColor: '#f0f9ff',
-            borderRadius: '6px',
-            marginBottom: '16px',
-            textAlign: 'center',
-            fontSize: '16px',
-            color: '#1e40af',
-          }}
-        >
-          {iterationStatus === 'pending' && 'Queued...'}
-          {iterationStatus === 'analyzing' && 'Analyzing repository and generating proposals...'}
-          {iterationStatus === 'implementing' &&
-            'Implementing all proposals... This may take a few minutes.'}
-        </div>
-      )}
+      {/* Progress indicator with log streaming */}
+      {isInProgress && <LogPanel logState={logStreamState} />}
 
       {/* Completed: show proposals */}
       {iterationStatus === 'completed' && latestIteration && (

From 65c21bca17330a6212a92ec3d37fc79d4fea906f Mon Sep 17 00:00:00 2001
From: yu-sugimoto <yuyuyuk2002@gmail.com>
Date: Wed, 4 Mar 2026 11:46:38 +0900
Subject: [PATCH 2/5] =?UTF-8?q?feat:=20=E3=83=AD=E3=82=B0=E3=83=91?=
 =?UTF-8?q?=E3=83=8D=E3=83=AB=E3=81=AB=E3=82=B5=E3=83=B3=E3=83=89=E3=83=9C?=
 =?UTF-8?q?=E3=83=83=E3=82=AF=E3=82=B9=E4=BD=9C=E6=88=90=E3=81=AE=E3=83=A9?=
 =?UTF-8?q?=E3=83=99=E3=83=AB=E3=82=92=E8=BF=BD=E5=8A=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/app/service/k8s_service.py   |  5 ++
 docker/worker-analyze.py             |  7 ---
 docker/worker-createpr.py            | 71 +---------------------------
 docker/worker-implement.py           |  5 --
 frontend/src/components/LogPanel.tsx |  1 +
 5 files changed, 7 insertions(+), 82 deletions(-)

diff --git a/backend/app/service/k8s_service.py b/backend/app/service/k8s_service.py
index d786873..37595f8 100644
--- a/backend/app/service/k8s_service.py
+++ b/backend/app/service/k8s_service.py
@@ -477,6 +477,7 @@ async def stream_pod_logs(
 
         # 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(
@@ -489,6 +490,10 @@ async def stream_pod_logs(
                     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)
diff --git a/docker/worker-analyze.py b/docker/worker-analyze.py
index 00fbd59..f8bcfe7 100644
--- a/docker/worker-analyze.py
+++ b/docker/worker-analyze.py
@@ -10,7 +10,6 @@
 import re
 import subprocess
 import sys
-import time
 from pathlib import Path
 
 import boto3
@@ -23,7 +22,6 @@ def emit_log(phase: str, message: str, detail: str | None = None) -> None:
     entry = {
         "phase": phase,
         "message": message,
-        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
     }
     if detail:
         entry["detail"] = detail
@@ -142,7 +140,6 @@ async def take_before_screenshot(repo_dir: str, screenshot_output: str) -> None:
                 if content_block.get("type") == "tool_use":
                     current_tool = content_block.get("name")
                     tool_input_chunks = ""
-                    emit_log("screenshot", f"Tool: {current_tool}")
             elif etype == "content_block_delta":
                 delta = event.get("delta", {})
                 if delta.get("type") == "input_json_delta":
@@ -158,10 +155,8 @@ async def take_before_screenshot(repo_dir: str, screenshot_output: str) -> None:
             for block in msg.content:
                 if isinstance(block, TextBlock):
                     emit_log("screenshot", block.text[:200], detail=block.text)
-                    print(f"Screenshot Agent: {block.text[:200]}")
                 elif isinstance(block, ToolUseBlock):
                     _emit_tool_detail("screenshot", block.name, json.dumps(block.input))
-                    print(f"Screenshot Tool: {block.name}")
 
 
 def _extract_proposals_json(collected_text: list[str]) -> list[dict]:
@@ -291,7 +286,6 @@ async def generate_proposals(
                 if content_block.get("type") == "tool_use":
                     current_tool = content_block.get("name")
                     tool_input_chunks = ""
-                    emit_log("analyzing", f"Tool: {current_tool}")
             elif etype == "content_block_delta":
                 delta = event.get("delta", {})
                 if delta.get("type") == "input_json_delta":
@@ -311,7 +305,6 @@ async def generate_proposals(
                     collected_text.append(block.text)
                 elif isinstance(block, ToolUseBlock):
                     _emit_tool_detail("analyzing", block.name, json.dumps(block.input))
-                    print(f"Analyze Tool: {block.name}")
 
     return _extract_proposals_json(collected_text)
 
diff --git a/docker/worker-createpr.py b/docker/worker-createpr.py
index b428d8e..954b6c0 100644
--- a/docker/worker-createpr.py
+++ b/docker/worker-createpr.py
@@ -10,46 +10,12 @@
 import re
 import subprocess
 import sys
-import time
 from datetime import datetime, timezone
 from pathlib import Path
 
 import boto3
 from botocore.config import Config
-from claude_agent_sdk import ClaudeAgentOptions, query, AssistantMessage, TextBlock, ToolUseBlock
-from claude_agent_sdk.types import StreamEvent
-
-
-def emit_log(phase: str, message: str, detail: str | None = None) -> None:
-    entry = {
-        "phase": phase,
-        "message": message,
-        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-    }
-    if detail:
-        entry["detail"] = detail
-    print(f"@@LOG@@{json.dumps(entry, ensure_ascii=False)}", flush=True)
-
-
-def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
-    """Parse tool input JSON and emit a human-readable log line."""
-    try:
-        params = json.loads(raw_input)
-    except json.JSONDecodeError:
-        emit_log(phase, f"Using tool: {tool_name}")
-        return
-
-    if tool_name == "Read":
-        emit_log(phase, f"Reading: {params.get('file_path', '?')}")
-    elif tool_name in ("Write", "Edit"):
-        emit_log(phase, f"Editing: {params.get('file_path', '?')}")
-    elif tool_name == "Bash":
-        cmd = params.get("command", "?")
-        emit_log(phase, f"Running: {cmd[:100]}")
-    elif tool_name in ("Glob", "Grep"):
-        emit_log(phase, f"Searching: {params.get('pattern', '?')}")
-    else:
-        emit_log(phase, f"Using tool: {tool_name}")
+from claude_agent_sdk import ClaudeAgentOptions, query, AssistantMessage, TextBlock
 
 
 def get_s3_client():
@@ -122,46 +88,18 @@ async def push_and_create_pr(
 IMPORTANT: You MUST output the PR URL in that exact format so it can be extracted.
 """
     collected_text: list[str] = []
-    current_tool = None
-    tool_input_chunks = ""
 
     async for msg in query(
         prompt=prompt,
         options=ClaudeAgentOptions(
             allowed_tools=["Read", "Bash", "Glob", "Grep"],
             cwd=repo_dir, max_turns=15, max_budget_usd=1.0,
-            include_partial_messages=True,
         ),
     ):
-        if isinstance(msg, StreamEvent):
-            event = msg.event
-            etype = event.get("type")
-            if etype == "content_block_start":
-                content_block = event.get("content_block", {})
-                if content_block.get("type") == "tool_use":
-                    current_tool = content_block.get("name")
-                    tool_input_chunks = ""
-                    emit_log("creating_pr", f"Tool: {current_tool}")
-            elif etype == "content_block_delta":
-                delta = event.get("delta", {})
-                if delta.get("type") == "input_json_delta":
-                    tool_input_chunks += delta.get("partial_json", "")
-            elif etype == "content_block_stop":
-                if current_tool and tool_input_chunks:
-                    _emit_tool_detail("creating_pr", current_tool, tool_input_chunks)
-                current_tool = None
-                tool_input_chunks = ""
-            continue
-
         if isinstance(msg, AssistantMessage):
             for block in msg.content:
                 if isinstance(block, TextBlock):
                     collected_text.append(block.text)
-                    emit_log("creating_pr", block.text[:200], detail=block.text)
-                    print(f"Agent: {block.text[:200]}")
-                elif isinstance(block, ToolUseBlock):
-                    _emit_tool_detail("creating_pr", block.name, json.dumps(block.input))
-                    print(f"Agent Tool: {block.name}")
 
     full_text = "\n".join(collected_text)
     for line in full_text.split("\n"):
@@ -199,10 +137,8 @@ async def main() -> None:
         print(f"Error: Patch not found at s3://{bucket}/{patch_key}", file=sys.stderr)
         sys.exit(1)
     diff_content = Path(local_patch).read_text()
-    emit_log("cloning", f"Patch loaded ({len(diff_content)} chars)")
 
     # Step 2: Clone repository (with token auth for push)
-    emit_log("cloning", "Cloning repository")
     repo_dir = "/workspace/repo"
     if repo_url.startswith("https://github.com/"):
         auth_repo_url = repo_url.replace(
@@ -220,11 +156,9 @@ async def main() -> None:
     # Step 3: Create a new branch
     ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
     branch_name = f"feat/{ts}-ui-session-{session_id[:8]}"
-    emit_log("patching", f"Creating branch: {branch_name}")
     subprocess.run(["git", "checkout", "-b", branch_name], cwd=repo_dir, check=True)
 
     # Step 4: Apply the patch
-    emit_log("patching", "Applying patch")
     result = subprocess.run(
         ["git", "am", "--3way", local_patch],
         cwd=repo_dir, capture_output=True, text=True,
@@ -244,7 +178,6 @@ async def main() -> None:
     subprocess.run(["git", "fetch", "--unshallow"], cwd=repo_dir, capture_output=True)
 
     # Step 6: Push and create PR via Agent SDK
-    emit_log("pushing", "Pushing branch and creating PR")
     diff_summary = diff_content[:10000]
     pr_url = await push_and_create_pr(repo_dir, branch_name, branch, diff_summary)
 
@@ -259,8 +192,6 @@ async def main() -> None:
         print("Error: Failed to extract PR URL from agent output", file=sys.stderr)
         sys.exit(1)
 
-    emit_log("completed", "PR creation complete")
-
 
 if __name__ == "__main__":
     asyncio.run(main())
diff --git a/docker/worker-implement.py b/docker/worker-implement.py
index 17779cf..0d68c67 100644
--- a/docker/worker-implement.py
+++ b/docker/worker-implement.py
@@ -10,7 +10,6 @@
 import os
 import subprocess
 import sys
-import time
 from pathlib import Path
 
 import boto3
@@ -23,7 +22,6 @@ def emit_log(phase: str, message: str, detail: str | None = None) -> None:
     entry = {
         "phase": phase,
         "message": message,
-        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
     }
     if detail:
         entry["detail"] = detail
@@ -157,7 +155,6 @@ async def implement_proposal(
                 if content_block.get("type") == "tool_use":
                     current_tool = content_block.get("name")
                     tool_input_chunks = ""
-                    emit_log("implementing", f"Tool: {current_tool}")
             elif etype == "content_block_delta":
                 delta = event.get("delta", {})
                 if delta.get("type") == "input_json_delta":
@@ -174,10 +171,8 @@ async def implement_proposal(
             for block in msg.content:
                 if isinstance(block, TextBlock):
                     emit_log("implementing", block.text[:200], detail=block.text)
-                    print(f"Agent: {block.text[:200]}")
                 elif isinstance(block, ToolUseBlock):
                     _emit_tool_detail("implementing", block.name, json.dumps(block.input))
-                    print(f"Agent Tool: {block.name}")
 
 
 async def main() -> None:
diff --git a/frontend/src/components/LogPanel.tsx b/frontend/src/components/LogPanel.tsx
index bd728b6..875420a 100644
--- a/frontend/src/components/LogPanel.tsx
+++ b/frontend/src/components/LogPanel.tsx
@@ -2,6 +2,7 @@ import { useState, useRef, useEffect, useCallback } from 'react'
 import type { LogStreamState, JobLogState, LogEntry } from '../hooks/useLogStream'
 
 const PHASE_LABELS: Record<string, string> = {
+  sandbox: 'Creating Sandbox',
   cloning: 'Cloning',
   patching: 'Applying Patch',
   screenshot: 'Taking Screenshot',

From 399c9be7027efcbdabd340a1d1d2a6345714a73d Mon Sep 17 00:00:00 2001
From: yu-sugimoto <yuyuyuk2002@gmail.com>
Date: Wed, 4 Mar 2026 12:02:59 +0900
Subject: [PATCH 3/5] =?UTF-8?q?fix:=20=E3=83=84=E3=83=BC=E3=83=AB=E4=BD=BF?=
 =?UTF-8?q?=E7=94=A8=E6=99=82=E3=81=AE=E3=83=AD=E3=82=B0=E5=87=BA=E5=8A=9B?=
 =?UTF-8?q?=E3=82=92=E5=89=8A=E9=99=A4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docker/worker-analyze.py   | 3 ---
 docker/worker-implement.py | 3 ---
 2 files changed, 6 deletions(-)

diff --git a/docker/worker-analyze.py b/docker/worker-analyze.py
index f8bcfe7..29dff0f 100644
--- a/docker/worker-analyze.py
+++ b/docker/worker-analyze.py
@@ -33,7 +33,6 @@ def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
     try:
         params = json.loads(raw_input)
     except json.JSONDecodeError:
-        emit_log(phase, f"Using tool: {tool_name}")
         return
 
     if tool_name == "Read":
@@ -45,8 +44,6 @@ def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
         emit_log(phase, f"Running: {cmd[:100]}")
     elif tool_name in ("Glob", "Grep"):
         emit_log(phase, f"Searching: {params.get('pattern', '?')}")
-    else:
-        emit_log(phase, f"Using tool: {tool_name}")
 
 
 def get_s3_client():
diff --git a/docker/worker-implement.py b/docker/worker-implement.py
index 0d68c67..bb7ffb8 100644
--- a/docker/worker-implement.py
+++ b/docker/worker-implement.py
@@ -33,7 +33,6 @@ def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
     try:
         params = json.loads(raw_input)
     except json.JSONDecodeError:
-        emit_log(phase, f"Using tool: {tool_name}")
         return
 
     if tool_name == "Read":
@@ -45,8 +44,6 @@ def _emit_tool_detail(phase: str, tool_name: str, raw_input: str) -> None:
         emit_log(phase, f"Running: {cmd[:100]}")
     elif tool_name in ("Glob", "Grep"):
         emit_log(phase, f"Searching: {params.get('pattern', '?')}")
-    else:
-        emit_log(phase, f"Using tool: {tool_name}")
 
 
 def get_s3_client():

From 256d2f0a6b7f335ce69b5c0f977a7dc317aadf7a Mon Sep 17 00:00:00 2001
From: yu-sugimoto <yuyuyuk2002@gmail.com>
Date: Wed, 4 Mar 2026 12:14:19 +0900
Subject: [PATCH 4/5] =?UTF-8?q?feat:=20=E6=8F=90=E6=A1=88=E5=BE=8C?=
 =?UTF-8?q?=E3=82=82LogPanel=E3=82=92=E8=A1=A8=E7=A4=BA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 frontend/src/components/LogPanel.tsx | 12 ++++++++++--
 frontend/src/pages/SessionDetail.tsx |  4 +++-
 2 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/frontend/src/components/LogPanel.tsx b/frontend/src/components/LogPanel.tsx
index 875420a..1b70888 100644
--- a/frontend/src/components/LogPanel.tsx
+++ b/frontend/src/components/LogPanel.tsx
@@ -132,12 +132,20 @@ function LogViewer({ entries }: LogViewerProps) {
 
 interface LogPanelProps {
   logState: LogStreamState
+  defaultCollapsed?: boolean
 }
 
-export default function LogPanel({ logState }: LogPanelProps) {
-  const [expanded, setExpanded] = useState(false)
+export default function LogPanel({ logState, defaultCollapsed }: LogPanelProps) {
+  const [expanded, setExpanded] = useState(!defaultCollapsed)
   const [activeTab, setActiveTab] = useState<string | null>(null)
 
+  // 完了時に自動的に折りたたむ
+  useEffect(() => {
+    if (defaultCollapsed) {
+      setExpanded(false)
+    }
+  }, [defaultCollapsed])
+
   const jobKeys = [...logState.jobs.keys()]
 
   // Auto-select first tab
diff --git a/frontend/src/pages/SessionDetail.tsx b/frontend/src/pages/SessionDetail.tsx
index fa8901f..ae7453e 100644
--- a/frontend/src/pages/SessionDetail.tsx
+++ b/frontend/src/pages/SessionDetail.tsx
@@ -235,7 +235,9 @@ export default function SessionDetail() {
       )}
 
       {/* Progress indicator with log streaming */}
-      {isInProgress && <LogPanel logState={logStreamState} />}
+      {(isInProgress || logStreamState.jobs.size > 0) && (
+        <LogPanel logState={logStreamState} defaultCollapsed={!isInProgress} />
+      )}
 
       {/* Completed: show proposals */}
       {iterationStatus === 'completed' && latestIteration && (

From 55e69c6e828f23ae664c86f8acf3e4f9bfd7d9ea Mon Sep 17 00:00:00 2001
From: yu-sugimoto <yuyuyuk2002@gmail.com>
Date: Wed, 4 Mar 2026 12:25:38 +0900
Subject: [PATCH 5/5] =?UTF-8?q?feat:=20=E3=83=AD=E3=82=B0=E5=87=BA?=
 =?UTF-8?q?=E5=8A=9B=E3=81=AE=E6=94=B9=E5=96=84=E3=81=A8=E3=82=A8=E3=83=BC?=
 =?UTF-8?q?=E3=82=B8=E3=82=A7=E3=83=B3=E3=83=88=E3=81=AE=E7=8A=B6=E6=85=8B?=
 =?UTF-8?q?=E7=AE=A1=E7=90=86=E3=82=92=E5=BC=B7=E5=8C=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/app/service/k8s_service.py        |  3 +-
 backend/app/service/log_stream_service.py | 39 ++++++++-------
 frontend/src/components/LogPanel.tsx      | 58 +++++++++++------------
 frontend/src/hooks/useLogStream.ts        | 18 +++++--
 frontend/src/pages/SessionDetail.tsx      |  4 +-
 5 files changed, 68 insertions(+), 54 deletions(-)

diff --git a/backend/app/service/k8s_service.py b/backend/app/service/k8s_service.py
index 37595f8..ed805ba 100644
--- a/backend/app/service/k8s_service.py
+++ b/backend/app/service/k8s_service.py
@@ -500,7 +500,8 @@ async def stream_pod_logs(
             elapsed += poll_interval
 
         if not pod_name:
-            yield f"@@LOG@@{{\"phase\":\"waiting\",\"message\":\"Pod not found for job {job_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
diff --git a/backend/app/service/log_stream_service.py b/backend/app/service/log_stream_service.py
index 3785929..b520a43 100644
--- a/backend/app/service/log_stream_service.py
+++ b/backend/app/service/log_stream_service.py
@@ -28,9 +28,10 @@ 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):]
+        json_str = line[len(LOG_PREFIX) :]
         try:
-            return json.loads(json_str)
+            result: dict = json.loads(json_str)
+            return result
         except json.JSONDecodeError:
             logger.debug("Failed to parse log JSON: %s", json_str[:100])
             return None
@@ -63,9 +64,7 @@ def register_job(
             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
-        )
+        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."""
@@ -87,7 +86,7 @@ async def stream_session_logs(
         k8s = K8sService()
         queue: asyncio.Queue[dict | None] = asyncio.Queue()
         tracked_jobs: set[str] = set()
-        tasks: list[asyncio.Task] = []  # type: ignore[type-arg]
+        tasks: list[asyncio.Task[None]] = []
 
         async def _stream_single_job(job_info: JobInfo) -> None:
             """Stream logs from a single job into the shared queue."""
@@ -105,22 +104,26 @@ async def _stream_single_job(job_info: JobInfo) -> None:
                         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,
-                        })
+                        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),
-                })
+                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."""
diff --git a/frontend/src/components/LogPanel.tsx b/frontend/src/components/LogPanel.tsx
index 1b70888..e738a79 100644
--- a/frontend/src/components/LogPanel.tsx
+++ b/frontend/src/components/LogPanel.tsx
@@ -1,4 +1,4 @@
-import { useState, useRef, useEffect, useCallback } from 'react'
+import { useState, useRef, useEffect, useCallback, useMemo } from 'react'
 import type { LogStreamState, JobLogState, LogEntry } from '../hooks/useLogStream'
 
 const PHASE_LABELS: Record<string, string> = {
@@ -45,7 +45,8 @@ function getOverallPhase(logState: LogStreamState): string {
     latestPhase = job.phase
   }
   // If all are completed
-  const allCompleted = logState.jobs.size > 0 && [...logState.jobs.values()].every((j) => j.phase === 'completed')
+  const allCompleted =
+    logState.jobs.size > 0 && [...logState.jobs.values()].every((j) => j.phase === 'completed')
   if (allCompleted) return 'completed'
   return latestPhase
 }
@@ -81,7 +82,10 @@ function LogViewer({ entries }: LogViewerProps) {
   }, [])
 
   const startIndex = Math.max(0, Math.floor(scrollTop / ROW_HEIGHT) - 5)
-  const endIndex = Math.min(entries.length, Math.ceil((scrollTop + containerHeight) / ROW_HEIGHT) + 5)
+  const endIndex = Math.min(
+    entries.length,
+    Math.ceil((scrollTop + containerHeight) / ROW_HEIGHT) + 5,
+  )
   const visibleEntries = entries.slice(startIndex, endIndex)
 
   return (
@@ -101,9 +105,7 @@ function LogViewer({ entries }: LogViewerProps) {
       <div style={{ height: `${totalHeight}px`, position: 'relative' }}>
         {visibleEntries.map((entry, i) => {
           const index = startIndex + i
-          const time = entry.timestamp
-            ? new Date(entry.timestamp).toLocaleTimeString()
-            : ''
+          const time = entry.timestamp ? new Date(entry.timestamp).toLocaleTimeString() : ''
           return (
             <div
               key={index}
@@ -137,36 +139,37 @@ interface LogPanelProps {
 
 export default function LogPanel({ logState, defaultCollapsed }: LogPanelProps) {
   const [expanded, setExpanded] = useState(!defaultCollapsed)
-  const [activeTab, setActiveTab] = useState<string | null>(null)
+  const [prevDefaultCollapsed, setPrevDefaultCollapsed] = useState(defaultCollapsed)
+  const [selectedTab, setSelectedTab] = useState<string | null>(null)
 
-  // 完了時に自動的に折りたたむ
-  useEffect(() => {
+  // 完了時に自動的に折りたたむ（defaultCollapsed が false→true に変わった時のみ）
+  if (defaultCollapsed !== prevDefaultCollapsed) {
+    setPrevDefaultCollapsed(defaultCollapsed)
     if (defaultCollapsed) {
       setExpanded(false)
     }
-  }, [defaultCollapsed])
+  }
 
-  const jobKeys = [...logState.jobs.keys()]
+  const jobKeys = useMemo(() => [...logState.jobs.keys()], [logState.jobs])
 
-  // Auto-select first tab
-  useEffect(() => {
-    if (activeTab === null && jobKeys.length > 0) {
-      setActiveTab(jobKeys[0])
-    }
-  }, [jobKeys.length, activeTab, jobKeys])
+  // Derive activeTab: use selected if valid, otherwise first available
+  const activeTab =
+    selectedTab !== null && jobKeys.includes(selectedTab)
+      ? selectedTab
+      : jobKeys.length > 0
+        ? jobKeys[0]
+        : null
+  const setActiveTab = setSelectedTab
 
   const overallPhase = getOverallPhase(logState)
   const phaseLabel = getPhaseLabel(overallPhase)
 
-  const activeJobState: JobLogState | undefined =
-    activeTab ? logState.jobs.get(activeTab) : undefined
+  const activeJobState: JobLogState | undefined = activeTab
+    ? logState.jobs.get(activeTab)
+    : undefined
 
   const dotColor =
-    overallPhase === 'completed'
-      ? '#34d399'
-      : overallPhase === 'error'
-        ? '#f87171'
-        : '#60a5fa'
+    overallPhase === 'completed' ? '#34d399' : overallPhase === 'error' ? '#f87171' : '#60a5fa'
 
   return (
     <div
@@ -212,9 +215,7 @@ export default function LogPanel({ logState, defaultCollapsed }: LogPanelProps)
           {phaseLabel}
           {logState.isStreaming && overallPhase !== 'completed' && '...'}
         </span>
-        <span style={{ color: '#6e7681', fontSize: '12px' }}>
-          {expanded ? 'Hide' : 'Details'}
-        </span>
+        <span style={{ color: '#6e7681', fontSize: '12px' }}>{expanded ? 'Hide' : 'Details'}</span>
       </button>
 
       {/* Expanded content */}
@@ -237,8 +238,7 @@ export default function LogPanel({ logState, defaultCollapsed }: LogPanelProps)
                   style={{
                     padding: '8px 16px',
                     border: 'none',
-                    borderBottom:
-                      activeTab === key ? '2px solid #58a6ff' : '2px solid transparent',
+                    borderBottom: activeTab === key ? '2px solid #58a6ff' : '2px solid transparent',
                     backgroundColor: 'transparent',
                     color: activeTab === key ? '#c9d1d9' : '#6e7681',
                     cursor: 'pointer',
diff --git a/frontend/src/hooks/useLogStream.ts b/frontend/src/hooks/useLogStream.ts
index 4b09a84..5615e54 100644
--- a/frontend/src/hooks/useLogStream.ts
+++ b/frontend/src/hooks/useLogStream.ts
@@ -44,7 +44,9 @@ export function useLogStream(sessionId: string | null, enabled: boolean): LogStr
   const [reconnectCount, setReconnectCount] = useState(0)
 
   // Keep enabledRef in sync
-  enabledRef.current = enabled
+  useEffect(() => {
+    enabledRef.current = enabled
+  }, [enabled])
 
   const cleanup = useCallback(() => {
     if (eventSourceRef.current) {
@@ -72,16 +74,22 @@ export function useLogStream(sessionId: string | null, enabled: boolean): LogStr
     const isReconnect = connectedOnceRef.current
     if (!isReconnect) {
       jobsRef.current = new Map()
-      setState({ jobs: new Map(), isStreaming: true, isDone: false })
-    } else {
-      setState((prev) => ({ ...prev, isStreaming: true, isDone: false }))
     }
+    connectedOnceRef.current = true
+
+    // Set streaming state asynchronously via microtask to avoid synchronous setState in effect
+    queueMicrotask(() => {
+      if (!isReconnect) {
+        setState({ jobs: new Map(), isStreaming: true, isDone: false })
+      } else {
+        setState((prev) => ({ ...prev, isStreaming: true, isDone: false }))
+      }
+    })
 
     // On reconnect (2nd+ connection), skip old logs by requesting only recent ones
     const url = isReconnect
       ? `/api/sessions/${sessionId}/logs/stream?since_seconds=1`
       : `/api/sessions/${sessionId}/logs/stream`
-    connectedOnceRef.current = true
 
     const es = new EventSource(url)
     eventSourceRef.current = es
diff --git a/frontend/src/pages/SessionDetail.tsx b/frontend/src/pages/SessionDetail.tsx
index ae7453e..eb0dc20 100644
--- a/frontend/src/pages/SessionDetail.tsx
+++ b/frontend/src/pages/SessionDetail.tsx
@@ -20,7 +20,9 @@ export default function SessionDetail() {
       ? session.iterations[session.iterations.length - 1]
       : null
   const iterationStatusForHook = latestIterationForHook?.status ?? 'pending'
-  const isInProgressForHook = ['pending', 'analyzing', 'implementing'].includes(iterationStatusForHook)
+  const isInProgressForHook = ['pending', 'analyzing', 'implementing'].includes(
+    iterationStatusForHook,
+  )
   const logStreamState = useLogStream(sessionId ?? null, isInProgressForHook)
 
   const [selectedIndex, setSelectedIndex] = useState<number | null>(null)