レポートダッシュボード実装を5フェーズに分割：scaffold→core→report→配線→E2Eで9タスクを並行化する

データ集計の結果を可視化するレポートダッシュボードを新規実装する案件で、最初は「1 PR で全部やる」つもりでしたが、見積もりが膨らみすぎたので 5 フェーズ・9 タスク に分割しました。

分割の設計と、並行化を効かせるための「インターフェース凍結」のタイミングについてまとめます。

なぜ分割したか

最初の見積もりは「2 週間で 1 PR」でした。レビュー観点で考えると以下が問題です。

• 変更ファイル 30+、レビュアーが追えない
• どこかで詰まると 2 週間がそのまま遅れる
• 並行作業ができず、自分が手を止めると全停止
• テストもまとめて書くので「実装が動かないとテストも動かない」状態

「縦に長い 1 PR」より「横に並列な複数 PR」のほうが、AI エージェントとの分業も含めて効率がよいと判断しました。

5 フェーズの構造

Phase 1 の重要性: ここでインターフェースを凍結します。Phase 2〜5 は Phase 1 で定義した型に依存して動くので、Phase 1 さえ確定すれば残りは並行可能になります。

Phase 1: Scaffold（最重要）

やること

1
from dataclasses import dataclass
2
from datetime import date
3

4
@dataclass(frozen=True)
5
class Summary:
6
    target_date: date
7
    total_count: int
8
    breakdown: dict[str, int]
9
    aggregated_metrics: dict[str, float]
10

11
# src/dashboard/aggregator.py
12
def aggregate(records: list[dict]) -> Summary:
13
    raise NotImplementedError
14

7 collapsed lines
15
# src/dashboard/report.py
16
def render(summary: Summary) -> bytes:
17
    raise NotImplementedError
18

19
# tests/test_aggregator.py
20
def test_aggregate_empty():
21
    pytest.skip("not implemented")

ポイント: 中身は空でも、型と関数シグネチャ・モジュール境界・テストファイル名を全部置く。Phase 2〜5 はこれを import して開発する。

Phase 1 で決めるべきこと

• データソースの型（list[dict] か Pandas か Polars か）
• 中間データの型（Summary の構造）
• 出力形式（bytes、ファイルパス、ストリーム）
• 例外の型（AggregationError、RenderError など）
• ログのプレフィックス（[dashboard]）

これが全タスク間の 共有合意 になります。Phase 1 が中途半端だと、Phase 2 と Phase 3 で同じ型を別の名前で使い始めて、後で大きな修正が発生します。

Phase 2: Core（並行可能）

集計ロジックの実装です。Phase 1 で aggregate(records) -> Summary という契約が決まっているので、入力と出力さえ守れば中身は自由です。

1
def aggregate(records: list[dict]) -> Summary:
2
    if not records:
3
        return Summary(target_date=date.today(), total_count=0, ...)
4

5
    by_category = defaultdict(int)
6
    for r in records:
7
        by_category[r["category"]] += 1
8

9
    metrics = {
10
        "mean_value": statistics.mean(r["value"] for r in records),
11
        "p95_value": _percentile([r["value"] for r in records], 95),
12
    }
13
    return Summary(
14
        target_date=records[0]["date"],
15
        total_count=len(records),
3 collapsed lines
16
        breakdown=dict(by_category),
17
        aggregated_metrics=metrics,
18
    )

並行化のコツ: テストはこのフェーズと一緒に書きます。test_aggregator.py を実装と同じ PR で完結させると、レビュー時に「動くこと」が確認できます。

Phase 3: Report（Phase 2 と並行可能）

Phase 2 とまったく独立に進められます。Phase 1 で Summary の構造が確定しているので、ダミー Summary を作ってレンダリングだけ作れます。

1
def make_dummy_summary() -> Summary:
2
    return Summary(
3
        target_date=date(2026, 4, 30),
4
        total_count=42,
5
        breakdown={"A": 30, "B": 12},
6
        aggregated_metrics={"mean_value": 1.5, "p95_value": 3.2},
7
    )
8

9
# src/dashboard/report.py
10
def render(summary: Summary) -> bytes:
11
    template = env.get_template("daily.html")
12
    html = template.render(summary=summary)
13
    return html.encode("utf-8")

Phase 2 の実装が遅れていても、Phase 3 はダミーで先に進める。ここが並行化の効果が一番出る部分です。

Phase 4: Wiring（Phase 2・3 完了後）

スケジューラやデリバリへの配線です。Phase 2・3 の実装が動くようになってから着手します。

1
def main():
2
    records = fetch_records_for_today()
3
    summary = aggregate(records)
4
    output = render(summary)
5
    deliver(output, channel="dashboard")

1
<!-- com.example.dashboard.daily.plist -->
2
<key>ProgramArguments</key>
3
<array>
4
    <string>/usr/bin/env</string>
5
    <string>python</string>
6
    <string>-m</string>
7
    <string>dashboard.runner</string>
8
</array>
9
<key>StartCalendarInterval</key>
10
<dict>
11
    <key>Hour</key><integer>9</integer>
12
    <key>Minute</key><integer>0</integer>
13
</dict>

ここは「配線」なので新しいロジックは増えません。ただし環境変数や PYTHONPATH の罠が多いので、ドライラン手順をしっかり書きます。

Phase 5: E2E

最後に結合テストと運用テストです。

1
def test_end_to_end_with_real_data():
2
    """実データの一部で全パイプラインを通す"""
3
    records = load_fixture_records("sample_2026_04_30.json")
4
    summary = aggregate(records)
5
    output = render(summary)
6

7
    # golden file 比較
8
    expected = (FIXTURES / "expected_dashboard_2026_04_30.html").read_bytes()
9
    assert output == expected

1
# launchd 環境を模した状態で 1 回実行し、生成物を確認
2
PYTHONPATH=src python -m dashboard.runner
3
test -f /var/data/dashboard/daily_2026_04_30.html || exit 1

運用テストの意義: 「ユニットテストが通る」と「launchd で動く」は別問題です。環境変数・cwd・PYTHONPATH が違うので、ドライランで確認します。

9 タスクへの分解

5 フェーズを 9 タスク にブレークダウンしました。

並行可能な組み合わせ: T1 → (T2, T3, T4, T5, T6 を並行) → (T7, T8 を並行) → T9。

直列でやると 9 タスク × 平均 0.5 日 = 4.5 日。並行化すると クリティカルパスは 3〜3.5 日 に縮みます。

並行化が効いた仕掛け

1. インターフェースの早期凍結

Phase 1 で型を確定させたので、Phase 2 と Phase 3 はお互いの実装を待たずに進められました。型を後で変えると 全タスクの修正になるので、Phase 1 のレビューに時間を使うのが正解です。

2. ダミーデータの fixture を Phase 1 で作る

Phase 3 の render(summary) を作る人は、Phase 2 の aggregate の完成を待ちません。Phase 1 で作った make_dummy_summary() を使ってテストを回せます。

3. テストファイルを Phase 1 で空作成

test_aggregator.py、test_report.py といった空ファイルを Phase 1 で作成しておくと、テストを書く人が「どのファイルに書くか」で迷いません。

4. PR を細かく出す

各タスクが 1 PR ですぐマージ可能なサイズになります。レビュアーは「1 タスクのスコープ」だけ見ればよく、レビュー速度が上がります。

失敗から学んだこと

罠 1: Phase 1 を急ぐと後で爆発する

最初は Phase 1 を「1 時間で終わらせる」つもりでした。しかし型の議論を急ぐと、Phase 3 の途中で「Summary に start_date が要る」と気づき、Phase 1 に戻って全タスクの再修正が発生しました。

学び: Phase 1 だけは時間をかけて議論する。半日かけても問題ないです。後の並行作業が全部 Phase 1 に依存するので。

罠 2: 並行作業の競合

T3 と T5 が同じファイルを編集しようとすると、マージ時に競合します。Phase 1 で モジュール境界（どのファイルがどのタスクの責務か）まで決めておきました。

罠 3: テストの fixture 重複

T4（aggregate のテスト）と T9（E2E）で似た fixture を別々に作ると、後で乖離します。tests/fixtures.py を Phase 1 の段階で作り、共有先として明示しました。

まとめ

「1 PR で全部やる」アプローチは、規模が大きくなると 直列化の罠 に陥ります。

最初に少し時間をかけてインターフェースを凍結し、5 フェーズに分割すると、AI エージェントへの並行委譲も含めて開発時間を大きく縮められます。Phase 1 の議論をしっかりやることが、後の全タスクの効率を決めると感じました。