feat: 各コンテナ内のログをポーリング#8
Conversation
There was a problem hiding this comment.
Pull request overview
セッション実行中の各コンテナ（analyze/implement/createpr）のログをバックエンドで集約して SSE で配信し、フロントエンド側で進捗表示をログパネルに置き換えるための変更です。
Changes:
- フロントエンドに SSE(EventSource) ベースのログ購読 hook とログ表示 UI（LogPanel）を追加
- バックエンドにセッション単位のログ集約サービスと SSE エンドポイントを追加（K8s Pod ログをストリーミング）
- 各 worker が構造化ログ（
@@LOG@@{...}）を stdout に出力するよう変更し、ジョブ登録を workflow から行うように更新
Reviewed changes
Copilot reviewed 17 out of 18 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| frontend/src/pages/SessionDetail.tsx | 進捗表示を LogPanel + useLogStream に置き換え |
| frontend/src/hooks/useLogStream.ts | SSE でログを購読・再接続する hook を追加 |
| frontend/src/components/LogPanel.tsx | ジョブ別タブ + 仮想化表示のログパネル UI を追加 |
| docker/worker-implement.py | 構造化ログ出力・partial stream から tool 実行状況をログ化 |
| docker/worker-createpr.py | Agent 出力の収集方法を型ベースに変更（ログ出力抑制） |
| docker/worker-analyze.py | 構造化ログ出力・partial stream から tool 実行状況をログ化 |
| backend/uv.lock | sse-starlette をロックに追加 |
| backend/pyproject.toml | sse-starlette 依存追加、mypy ignore 対象に追加 |
| backend/migration/versions/c3d4e5f6g7h8_add_session_iteration_tables.py | PostgreSQL ENUM 作成方法を明示化（create_type を制御） |
| backend/app/workflow/session_implementation_graph.py | implement job を LogStreamService に登録 |
| backend/app/workflow/session_create_pr_graph.py | createpr job を LogStreamService に登録 |
| backend/app/workflow/session_analyzer_graph.py | analyze job を LogStreamService に登録 |
| backend/app/usecase/session_usecase.py | iteration status 更新を optimistic lock リトライに集約 |
| backend/app/service/log_stream_service.py | セッション内ジョブのログを多重化して配信するサービスを追加 |
| backend/app/service/k8s_service.py | K8s pod ログの async ストリーミング API を追加 |
| backend/app/router/sessions.py | セッションログ SSE エンドポイントを追加 |
| backend/app/di/dependencies.py | LogStreamService の DI dependency を追加 |
| backend/app/di/container.py | LogStreamService をコンテナで singleton 管理 |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| finally: | ||
| producer_task.cancel() | ||
|
|
There was a problem hiding this comment.
SSE クライアント切断時に producer_task.cancel() していますが、asyncio.to_thread(...) で動いているログ読み取りはキャンセルできず、バックグラウンドで読み続けてリソースを消費する可能性があります。切断時にレスポンス/ストリームを閉じる・停止フラグを渡す等で producer を確実に停止できるようにしてください。
| """ | ||
|
|
||
| import asyncio | ||
| import json |
There was a problem hiding this comment.
json を import していますが、このファイル内で参照がなく未使用です。不要であれば削除してください（lint/mypy 設定次第ではエラーになります）。
| import json |
| if (defaultCollapsed !== prevDefaultCollapsed) { | ||
| setPrevDefaultCollapsed(defaultCollapsed) | ||
| if (defaultCollapsed) { | ||
| setExpanded(false) | ||
| } | ||
| } |
There was a problem hiding this comment.
defaultCollapsed の変化を検知するために render 中に setPrevDefaultCollapsed / setExpanded を呼んでおり、React の "state update during render" になって無限再レンダーや警告の原因になります。useEffect で defaultCollapsed を監視して状態更新する形に移してください。
| if (defaultCollapsed !== prevDefaultCollapsed) { | |
| setPrevDefaultCollapsed(defaultCollapsed) | |
| if (defaultCollapsed) { | |
| setExpanded(false) | |
| } | |
| } | |
| useEffect(() => { | |
| if (defaultCollapsed !== prevDefaultCollapsed) { | |
| setPrevDefaultCollapsed(defaultCollapsed) | |
| if (defaultCollapsed) { | |
| setExpanded(false) | |
| } | |
| } | |
| }, [defaultCollapsed, prevDefaultCollapsed]) |
|
|
||
| useEffect(() => { | ||
| if (!sessionId || !enabled) { | ||
| cleanup() |
There was a problem hiding this comment.
enabled が false になった場合に cleanup() は呼ばれますが、isStreaming / isDone を含む state を更新していないため、接続が閉じた後も isStreaming: true のまま UI に残る可能性があります。!enabled / !sessionId の分岐でも state を適切に更新（例: isStreaming=false, 必要なら isDone=true）してください。
| cleanup() | |
| cleanup() | |
| setState((prev) => ({ | |
| ...prev, | |
| isStreaming: false, | |
| isDone: true, | |
| })) |
| def cleanup_session(self, session_id: str) -> None: | ||
| """Remove session data after streaming is done.""" | ||
| self._sessions.pop(session_id, None) |
There was a problem hiding this comment.
LogStreamService は session/job 情報を self._sessions に保持しますが、削除処理（cleanup_session）がどこからも呼ばれていないためプロセス内メモリが増え続けます。ストリーミング完了時やセッション完了時に必ず cleanup するか、TTL で自動的に expire する仕組みを入れてください。
|
|
||
| def __init__(self) -> None: | ||
| self._sessions: dict[str, SessionJobs] = {} | ||
| self._lock = asyncio.Lock() |
There was a problem hiding this comment.
self._lock = asyncio.Lock() が宣言されていますが、このファイル内で使用されておらずデッドコードになっています。ロックが必要なら register_job / stream_session_logs の _sessions アクセスを保護する用途で使うか、不要なら削除してください。
| self._lock = asyncio.Lock() |
| @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), |
There was a problem hiding this comment.
新しい SSE エンドポイント /api/sessions/{session_id}/logs/stream に対する API テストが追加されていません。既存で backend/tests/test_api.py が API を広くカバーしているので、get_log_stream_service をスタブして log/done イベントと since_seconds の挙動を検証するテストを追加してください。
No description provided.