yoloyolo/.planning/phases/01-capture-and-classification/01-03-PLAN.md

---
phase: 01-capture-and-classification
plan: 03
type: execute
wave: 2
depends_on: ["01-01"]
files_modified:
  - aggregate/window.go
  - aggregate/summary.go
  - aggregate/window_test.go
  - aggregate/summary_test.go
autonomous: true
requirements: [CLAS-03, CLAS-04]

must_haves:
  truths:
    - "On exit, user sees a per-protocol packet count summary printed to stderr"
    - "User can pass --verbose and see per-window protocol activity lines on stderr"
    - "Aggregator accumulates packet counts into time windows and emits WindowSnapshots"
  artifacts:
    - path: "aggregate/window.go"
      provides: "Ticker-driven window aggregator that emits WindowSnapshots"
      exports: ["Aggregate"]
    - path: "aggregate/summary.go"
      provides: "PrintSummary and PrintWindowLine functions for stderr output"
      exports: ["PrintSummary", "PrintWindowLine"]
    - path: "aggregate/window_test.go"
      provides: "Tests for window aggregation logic"
      min_lines: 50
    - path: "aggregate/summary_test.go"
      provides: "Tests for summary formatting and verbose output"
      min_lines: 60
  key_links:
    - from: "aggregate/window.go"
      to: "classify/types.go"
      via: "Consumes ClassifiedPacket, produces WindowSnapshot"
      pattern: "classify\\.ClassifiedPacket"
    - from: "aggregate/summary.go"
      to: "classify/types.go"
      via: "Uses TrafficClass and WindowSnapshot for formatting"
      pattern: "classify\\.TrafficClass"
---

<objective>
Implement the aggregation and stats reporting package: time-windowed accumulation, exit summary, and verbose per-window output.

Purpose: This package bridges classified packets and the eventual audio engine (Phase 2). It accumulates per-class counts into time windows, prints verbose activity lines during capture, and prints a final protocol summary on exit.

Output: aggregate/ package with Aggregate, PrintSummary, PrintWindowLine functions and full unit tests.
</objective>

<execution_context>
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/phases/01-capture-and-classification/01-CONTEXT.md
@.planning/phases/01-capture-and-classification/01-RESEARCH.md
@.planning/phases/01-capture-and-classification/01-01-SUMMARY.md

<interfaces>
<!-- From classify/types.go (created in Plan 01): -->
```go
package classify

type TrafficClass string

const (
    ClassICMP     TrafficClass = "ICMP"
    ClassDNS      TrafficClass = "DNS"
    ClassHTTPS    TrafficClass = "HTTPS"
    ClassHTTP     TrafficClass = "HTTP"
    ClassSSH      TrafficClass = "SSH"
    ClassSMTP     TrafficClass = "SMTP"
    ClassNTP      TrafficClass = "NTP"
    ClassDHCP     TrafficClass = "DHCP"
    ClassOtherTCP TrafficClass = "other-TCP"
    ClassOtherUDP TrafficClass = "other-UDP"
    ClassUnknown  TrafficClass = "unknown"
)

func AllClasses() []TrafficClass { ... }

type ClassifiedPacket struct {
    Class     TrafficClass
    SrcPort   uint16
    DstPort   uint16
    Protocol  string
    Length    int
}

type WindowSnapshot struct {
    Counts       map[TrafficClass]int64
    TotalPackets int64
    WindowIndex  int
}
```
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Implement ticker-driven window aggregator</name>
  <files>aggregate/window.go, aggregate/window_test.go</files>
  <read_first>
    classify/types.go
    .planning/phases/01-capture-and-classification/01-RESEARCH.md
  </read_first>
  <behavior>
    - TestAggregateEmitsSnapshot: Send 5 ClassifiedPackets, trigger window tick, receive WindowSnapshot with correct counts
    - TestAggregateMultipleClasses: Send packets of 3 different classes, verify snapshot has all 3 with correct counts
    - TestAggregateDoneFlushesPartial: Close done channel mid-window, verify final partial snapshot is emitted
    - TestAggregateEmptyWindow: Trigger tick with no packets, verify snapshot has zero counts and TotalPackets==0
    - TestAggregateWindowIndex: Multiple ticks increment WindowIndex (0, 1, 2...)
  </behavior>
  <action>
1. Create aggregate/window.go implementing the ticker-driven aggregation pattern from RESEARCH.md Pattern 2:

```go
package aggregate

import (
    "time"

    "github.com/netsynth/netsynth/classify"
)

// DefaultWindowMs is the default time window duration in milliseconds.
// Claude's discretion per CONTEXT.md.
const DefaultWindowMs = 500

// Aggregate reads ClassifiedPackets from events, accumulates counts per TrafficClass
// in time windows of windowMs duration, and emits a WindowSnapshot per window.
// When done is closed, flushes the current partial window and closes the output channel.
// If verbose is true, calls onSnapshot for each window (used for --verbose stderr output).
func Aggregate(done <-chan struct{}, events <-chan classify.ClassifiedPacket,
    windowMs int, onSnapshot func(classify.WindowSnapshot)) <-chan classify.WindowSnapshot {

    out := make(chan classify.WindowSnapshot, 8)
    go func() {
        defer close(out)
        ticker := time.NewTicker(time.Duration(windowMs) * time.Millisecond)
        defer ticker.Stop()

        counts := make(map[classify.TrafficClass]int64)
        var totalPackets int64
        windowIndex := 0

        flush := func() {
            snap := classify.WindowSnapshot{
                Counts:       counts,
                TotalPackets: totalPackets,
                WindowIndex:  windowIndex,
            }
            if onSnapshot != nil {
                onSnapshot(snap)
            }
            out <- snap
            counts = make(map[classify.TrafficClass]int64)
            totalPackets = 0
            windowIndex++
        }

        for {
            select {
            case <-done:
                flush() // flush final partial window
                return
            case <-ticker.C:
                flush()
            case ev, ok := <-events:
                if !ok {
                    flush()
                    return
                }
                counts[ev.Class]++
                totalPackets++
            }
        }
    }()
    return out
}
```

2. Create aggregate/window_test.go:
   - Use a done channel and a buffered events channel to control timing without real tickers
   - For ticker tests, use a short windowMs (10ms) and sleep briefly to trigger ticks
   - For done-flush tests, close done channel immediately and verify output
   - Verify WindowSnapshot.Counts map contains expected class keys and values
   - Verify WindowSnapshot.TotalPackets matches sum of counts
   - Verify WindowIndex increments across windows

Write tests first (RED), then implement (GREEN).
  </action>
  <verify>
    <automated>cd /home/dev/workspace/yoloyolo && go test -v -count=1 -race ./aggregate/...</automated>
  </verify>
  <acceptance_criteria>
    - aggregate/window.go contains `func Aggregate(done <-chan struct{}, events <-chan classify.ClassifiedPacket`
    - aggregate/window.go contains `const DefaultWindowMs = 500`
    - aggregate/window.go contains `time.NewTicker`
    - aggregate/window.go imports `github.com/netsynth/netsynth/classify`
    - aggregate/window_test.go contains `TestAggregate` with subtests for snapshot emission, multiple classes, done flush, empty window, and window index
    - `go test -v -count=1 -race ./aggregate/...` exits 0
  </acceptance_criteria>
  <done>Aggregate function emits WindowSnapshots on ticker ticks and flushes partial window on done. All timing-based tests pass including race detector.</done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: Implement exit summary and verbose window output formatting</name>
  <files>aggregate/summary.go, aggregate/summary_test.go</files>
  <read_first>
    classify/types.go
    aggregate/window.go
    .planning/phases/01-capture-and-classification/01-RESEARCH.md
  </read_first>
  <behavior>
    - TestPrintSummary: Given totals map {ICMP: 10, DNS: 50, HTTPS: 100}, output contains "ICMP", "DNS", "HTTPS" with correct counts and percentages, plus "TOTAL" line
    - TestPrintSummaryEmpty: Given empty totals, output contains "TOTAL" with 0 packets
    - TestPrintSummarySorted: Output classes appear in stable alphabetical or canonical order (not random map order)
    - TestPrintWindowLine: Given WindowSnapshot with {DNS: 5, HTTPS: 10}, output contains window index, class names, and counts on a single line
    - TestPrintWindowLineEmpty: Given empty snapshot, output contains window index with no class entries
    - TestAccumulateTotals: Given multiple WindowSnapshots, AccumulateTotals produces correct cumulative counts
  </behavior>
  <action>
1. Create aggregate/summary.go:

```go
package aggregate

import (
    "fmt"
    "io"
    "sort"

    "github.com/netsynth/netsynth/classify"
)

// PrintSummary writes a per-protocol packet count summary to w (typically os.Stderr).
// Output format matches tcpdump conventions (CLAS-03).
func PrintSummary(w io.Writer, totals map[classify.TrafficClass]int64) {
    var totalPackets int64
    for _, count := range totals {
        totalPackets += count
    }

    fmt.Fprintln(w, "\n--- Protocol Summary ---")

    // Sort classes for stable output
    classes := make([]string, 0, len(totals))
    for c := range totals {
        classes = append(classes, string(c))
    }
    sort.Strings(classes)

    for _, className := range classes {
        c := classify.TrafficClass(className)
        count := totals[c]
        pct := 0.0
        if totalPackets > 0 {
            pct = float64(count) / float64(totalPackets) * 100.0
        }
        fmt.Fprintf(w, "  %-15s %8d packets (%5.1f%%)\n", c, count, pct)
    }
    fmt.Fprintf(w, "  %-15s %8d packets\n", "TOTAL", totalPackets)
}

// PrintWindowLine writes a single verbose per-window activity line to w (CLAS-04).
// Format: [window N] CLASS:count CLASS:count ... (total: N)
func PrintWindowLine(w io.Writer, snap classify.WindowSnapshot) {
    fmt.Fprintf(w, "[window %d]", snap.WindowIndex)
    // Sort for stable output
    classes := make([]string, 0, len(snap.Counts))
    for c := range snap.Counts {
        classes = append(classes, string(c))
    }
    sort.Strings(classes)
    for _, className := range classes {
        c := classify.TrafficClass(className)
        if snap.Counts[c] > 0 {
            fmt.Fprintf(w, " %s:%d", c, snap.Counts[c])
        }
    }
    fmt.Fprintf(w, " (total: %d)\n", snap.TotalPackets)
}

// AccumulateTotals merges a WindowSnapshot's counts into cumulative totals.
func AccumulateTotals(totals map[classify.TrafficClass]int64, snap classify.WindowSnapshot) {
    for class, count := range snap.Counts {
        totals[class] += count
    }
}
```

2. Create aggregate/summary_test.go:
   - Use bytes.Buffer as io.Writer to capture output
   - TestPrintSummary: Verify output contains expected class names, counts, percentages, and TOTAL line
   - TestPrintSummaryEmpty: Verify output contains "TOTAL" with 0
   - TestPrintSummarySorted: Verify output lines are in alphabetical order (capture output, split lines, check ordering)
   - TestPrintWindowLine: Verify format "[window 0] DNS:5 HTTPS:10 (total: 15)"
   - TestAccumulateTotals: Two snapshots accumulated, verify merged counts

Write tests first (RED), then implement (GREEN).
  </action>
  <verify>
    <automated>cd /home/dev/workspace/yoloyolo && go test -v -count=1 ./aggregate/...</automated>
  </verify>
  <acceptance_criteria>
    - aggregate/summary.go contains `func PrintSummary(w io.Writer, totals map[classify.TrafficClass]int64)`
    - aggregate/summary.go contains `func PrintWindowLine(w io.Writer, snap classify.WindowSnapshot)`
    - aggregate/summary.go contains `func AccumulateTotals(totals map[classify.TrafficClass]int64, snap classify.WindowSnapshot)`
    - aggregate/summary.go contains `"--- Protocol Summary ---"`
    - aggregate/summary.go contains `sort.Strings` for stable output ordering
    - aggregate/summary_test.go contains `TestPrintSummary`, `TestPrintSummaryEmpty`, `TestPrintSummarySorted`
    - aggregate/summary_test.go contains `TestPrintWindowLine`, `TestAccumulateTotals`
    - `go test -v -count=1 ./aggregate/...` exits 0
  </acceptance_criteria>
  <done>PrintSummary outputs per-protocol counts with percentages and TOTAL to stderr (CLAS-03). PrintWindowLine outputs per-window activity for --verbose mode (CLAS-04). AccumulateTotals correctly merges snapshots. All formatting tests pass with stable sorted output.</done>
</task>

</tasks>

<verification>
- `go test -v -count=1 -race ./aggregate/...` all tests pass
- `go vet ./aggregate/...` no issues
- Output format matches tcpdump convention (stderr, per-protocol counts with percentages)
</verification>

<success_criteria>
- Aggregate emits WindowSnapshots on time ticks and flushes on done
- PrintSummary prints per-protocol packet counts with percentages and TOTAL to an io.Writer
- PrintWindowLine prints per-window activity for verbose mode
- AccumulateTotals correctly accumulates across multiple snapshots
- All output is sorted for deterministic display
- All unit tests passing (including race detector)
</success_criteria>

<output>
After completion, create `.planning/phases/01-capture-and-classification/01-03-SUMMARY.md`
</output>