diff --git a/server/cmd/api/api/api.go b/server/cmd/api/api/api.go
index 6e5b4440..4f6c671e 100644
--- a/server/cmd/api/api/api.go
+++ b/server/cmd/api/api/api.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"os"
 	"os/exec"
 	"sync"
@@ -103,7 +104,7 @@ func New(
 		return nil, fmt.Errorf("captureSession cannot be nil")
 	}
 
-	mon := cdpmonitor.New(upstreamMgr, captureSession.Publish, displayNum)
+	mon := cdpmonitor.New(upstreamMgr, captureSession.Publish, displayNum, slog.Default())
 	ctx, cancel := context.WithCancel(context.Background())
 
 	return &ApiService{
diff --git a/server/lib/cdpmonitor/cdp_test.go b/server/lib/cdpmonitor/cdp_test.go
new file mode 100644
index 00000000..0917451c
--- /dev/null
+++ b/server/lib/cdpmonitor/cdp_test.go
@@ -0,0 +1,371 @@
+package cdpmonitor
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"image"
+	"image/color"
+	"image/png"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
+	"github.com/kernel/kernel-images/server/lib/events"
+	"github.com/stretchr/testify/require"
+)
+
+var discardLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
+
+// minimalPNG is a valid 1x1 PNG used as a test fixture for screenshot tests.
+var minimalPNG = func() []byte {
+	img := image.NewRGBA(image.Rect(0, 0, 1, 1))
+	img.Set(0, 0, color.RGBA{R: 255, G: 0, B: 0, A: 255})
+	var buf bytes.Buffer
+	_ = png.Encode(&buf, img)
+	return buf.Bytes()
+}()
+
+// testServer is a minimal WebSocket server that accepts connections and
+// lets the test drive scripted message sequences.
+type testServer struct {
+	srv    *httptest.Server
+	conn   *websocket.Conn
+	connMu sync.Mutex
+	connCh chan struct{} // closed when the first connection is accepted
+	msgCh  chan []byte   // inbound messages from Monitor
+}
+
+func newTestServer(t *testing.T) *testServer {
+	t.Helper()
+	s := &testServer{
+		msgCh:  make(chan []byte, 128),
+		connCh: make(chan struct{}),
+	}
+	var connOnce sync.Once
+	s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true})
+		if err != nil {
+			return
+		}
+		s.connMu.Lock()
+		s.conn = c
+		s.connMu.Unlock()
+		connOnce.Do(func() { close(s.connCh) })
+		go func() {
+			for {
+				_, b, err := c.Read(context.Background())
+				if err != nil {
+					return
+				}
+				s.msgCh <- b
+			}
+		}()
+	}))
+	return s
+}
+
+func (s *testServer) wsURL() string {
+	return "ws" + strings.TrimPrefix(s.srv.URL, "http")
+}
+
+func (s *testServer) sendToMonitor(t *testing.T, msg any) {
+	t.Helper()
+	s.connMu.Lock()
+	c := s.conn
+	s.connMu.Unlock()
+	require.NotNil(t, c, "no active connection")
+	require.NoError(t, wsjson.Write(context.Background(), c, msg))
+}
+
+func (s *testServer) readFromMonitor(t *testing.T, timeout time.Duration) cdpMessage {
+	t.Helper()
+	select {
+	case b := <-s.msgCh:
+		var msg cdpMessage
+		require.NoError(t, json.Unmarshal(b, &msg))
+		return msg
+	case <-time.After(timeout):
+		t.Fatal("timeout waiting for message from Monitor")
+		return cdpMessage{}
+	}
+}
+
+func (s *testServer) close() {
+	s.connMu.Lock()
+	if s.conn != nil {
+		_ = s.conn.Close(websocket.StatusNormalClosure, "done")
+	}
+	s.connMu.Unlock()
+	s.srv.Close()
+}
+
+// testUpstream implements UpstreamProvider for tests.
+type testUpstream struct {
+	mu      sync.Mutex
+	current string
+	subs    []chan string
+}
+
+func newTestUpstream(url string) *testUpstream {
+	return &testUpstream{current: url}
+}
+
+func (u *testUpstream) Current() string {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	return u.current
+}
+
+func (u *testUpstream) Subscribe() (<-chan string, func()) {
+	ch := make(chan string, 1)
+	u.mu.Lock()
+	u.subs = append(u.subs, ch)
+	u.mu.Unlock()
+	cancel := func() {
+		u.mu.Lock()
+		for i, s := range u.subs {
+			if s == ch {
+				u.subs = append(u.subs[:i], u.subs[i+1:]...)
+				break
+			}
+		}
+		u.mu.Unlock()
+		close(ch)
+	}
+	return ch, cancel
+}
+
+func (u *testUpstream) notifyRestart(newURL string) {
+	u.mu.Lock()
+	u.current = newURL
+	subs := make([]chan string, len(u.subs))
+	copy(subs, u.subs)
+	u.mu.Unlock()
+	for _, ch := range subs {
+		select {
+		case ch <- newURL:
+		default:
+		}
+	}
+}
+
+// eventCollector captures published events with channel-based notification.
+type eventCollector struct {
+	mu     sync.Mutex
+	events []events.Event
+	notify chan struct{} // signaled on every publish
+}
+
+func newEventCollector() *eventCollector {
+	return &eventCollector{notify: make(chan struct{}, 256)}
+}
+
+func (c *eventCollector) publishFn() PublishFunc {
+	return func(ev events.Event) {
+		c.mu.Lock()
+		c.events = append(c.events, ev)
+		c.mu.Unlock()
+		select {
+		case c.notify <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// waitFor blocks until an event of the given type is published, or fails.
+func (c *eventCollector) waitFor(t *testing.T, eventType string, timeout time.Duration) events.Event {
+	t.Helper()
+	deadline := time.After(timeout)
+	for {
+		c.mu.Lock()
+		for _, ev := range c.events {
+			if ev.Type == eventType {
+				c.mu.Unlock()
+				return ev
+			}
+		}
+		c.mu.Unlock()
+		select {
+		case <-c.notify:
+		case <-deadline:
+			t.Fatalf("timeout waiting for event type=%q", eventType)
+			return events.Event{}
+		}
+	}
+}
+
+// waitForNew blocks until a NEW event of the given type is published after this
+// call, ignoring any events already in the collector.
+func (c *eventCollector) waitForNew(t *testing.T, eventType string, timeout time.Duration) events.Event {
+	t.Helper()
+	c.mu.Lock()
+	skip := len(c.events)
+	c.mu.Unlock()
+
+	deadline := time.After(timeout)
+	for {
+		c.mu.Lock()
+		for i := skip; i < len(c.events); i++ {
+			if c.events[i].Type == eventType {
+				ev := c.events[i]
+				c.mu.Unlock()
+				return ev
+			}
+		}
+		c.mu.Unlock()
+		select {
+		case <-c.notify:
+		case <-deadline:
+			t.Fatalf("timeout waiting for new event type=%q", eventType)
+			return events.Event{}
+		}
+	}
+}
+
+// assertNone verifies that no event of the given type arrives within d.
+func (c *eventCollector) assertNone(t *testing.T, eventType string, d time.Duration) {
+	t.Helper()
+	deadline := time.After(d)
+	for {
+		select {
+		case <-c.notify:
+			c.mu.Lock()
+			for _, ev := range c.events {
+				if ev.Type == eventType {
+					c.mu.Unlock()
+					t.Fatalf("unexpected event %q published", eventType)
+					return
+				}
+			}
+			c.mu.Unlock()
+		case <-deadline:
+			return
+		}
+	}
+}
+
+// ResponderFunc is called for each CDP command the Monitor sends.
+// Return nil to use the default empty result.
+type ResponderFunc func(msg cdpMessage) any
+
+// listenAndRespond drains srv.msgCh, calls fn for each command, and sends the
+// response. If fn is nil or returns nil, sends {"id": msg.ID, "result": {}}.
+func listenAndRespond(srv *testServer, stopCh <-chan struct{}, fn ResponderFunc) {
+	for {
+		select {
+		case b := <-srv.msgCh:
+			var msg cdpMessage
+			if json.Unmarshal(b, &msg) != nil || msg.ID == nil {
+				continue
+			}
+			srv.connMu.Lock()
+			c := srv.conn
+			srv.connMu.Unlock()
+			if c == nil {
+				continue
+			}
+			var resp any
+			if fn != nil {
+				resp = fn(msg)
+			}
+			if resp == nil {
+				resp = map[string]any{"id": msg.ID, "result": map[string]any{}}
+			}
+			_ = wsjson.Write(context.Background(), c, resp)
+		case <-stopCh:
+			return
+		}
+	}
+}
+
+// startMonitor creates a Monitor against srv, starts it, and returns a cleanup func.
+// Waits for Target.getTargets (the last command in initSession) before returning.
+func startMonitor(t *testing.T, srv *testServer, fn ResponderFunc) (*Monitor, *eventCollector, func()) {
+	t.Helper()
+	ec := newEventCollector()
+	upstream := newTestUpstream(srv.wsURL())
+	m := New(upstream, ec.publishFn(), 99, discardLogger)
+	require.NoError(t, m.Start(context.Background()))
+
+	// Closed when Target.getTargets is responded to (last command of initSession).
+	// Tests needing attachExistingTargets to finish should use require.Eventually.
+	initDone := make(chan struct{})
+	var initOnce sync.Once
+
+	wrappedFn := func(msg cdpMessage) any {
+		var result any
+		if fn != nil {
+			result = fn(msg)
+		}
+		if msg.Method == "Target.getTargets" {
+			initOnce.Do(func() { close(initDone) })
+		}
+		return result
+	}
+
+	stopResponder := make(chan struct{})
+	go listenAndRespond(srv, stopResponder, wrappedFn)
+
+	// Wait for the websocket connection to be established.
+	select {
+	case <-srv.connCh:
+	case <-time.After(3 * time.Second):
+		t.Fatal("fake server never received a connection")
+	}
+	// Wait for the init sequence to complete.
+	select {
+	case <-initDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("init sequence (Target.getTargets) did not complete")
+	}
+
+	cleanup := func() {
+		close(stopResponder)
+		m.Stop()
+	}
+	return m, ec, cleanup
+}
+
+// newComputedMonitor creates an unconnected Monitor for testing computed state
+// (network_idle, layout_settled, navigation_settled) without a real websocket.
+func newComputedMonitor(t *testing.T) (*Monitor, *eventCollector) {
+	t.Helper()
+	ec := newEventCollector()
+	upstream := newTestUpstream("ws://127.0.0.1:0")
+	m := New(upstream, ec.publishFn(), 0, discardLogger)
+	return m, ec
+}
+
+// navigateMonitor sends a Page.frameNavigated to reset computed state.
+func navigateMonitor(m *Monitor, url string) {
+	p, _ := json.Marshal(map[string]any{
+		"frame": map[string]any{"id": "f1", "url": url},
+	})
+	m.handleFrameNavigated(p, "s1")
+}
+
+// simulateRequest sends a Network.requestWillBeSent through the handler.
+func simulateRequest(m *Monitor, id string) {
+	p, _ := json.Marshal(map[string]any{
+		"requestId": id, "resourceType": "Document",
+		"request": map[string]any{"method": "GET", "url": "https://example.com/" + id},
+	})
+	m.handleNetworkRequest(p, "s1")
+}
+
+// simulateFinished stores minimal state and sends Network.loadingFinished.
+func simulateFinished(m *Monitor, id string) {
+	m.pendReqMu.Lock()
+	m.pendingRequests[id] = networkReqState{method: "GET", url: "https://example.com/" + id}
+	m.pendReqMu.Unlock()
+	p, _ := json.Marshal(map[string]any{"requestId": id})
+	m.handleLoadingFinished(p, "s1")
+}
diff --git a/server/lib/cdpmonitor/computed_test.go b/server/lib/cdpmonitor/computed_test.go
index 80801f3c..64295a83 100644
--- a/server/lib/cdpmonitor/computed_test.go
+++ b/server/lib/cdpmonitor/computed_test.go
@@ -1,7 +1,6 @@
 package cdpmonitor
 
 import (
-	"sync"
 	"testing"
 	"time"
 
@@ -9,81 +8,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-// eventCollector gathers published events for test assertions.
-type eventCollector struct {
-	mu   sync.Mutex
-	evs  []events.Event
-	subs []chan events.Event
-}
-
-func newEventCollector() *eventCollector { return &eventCollector{} }
-
-func (ec *eventCollector) publishFn() PublishFunc {
-	return func(ev events.Event) {
-		ec.mu.Lock()
-		ec.evs = append(ec.evs, ev)
-		for _, ch := range ec.subs {
-			select {
-			case ch <- ev:
-			default:
-			}
-		}
-		ec.mu.Unlock()
-	}
-}
-
-func (ec *eventCollector) subscribe() (<-chan events.Event, func()) {
-	ch := make(chan events.Event, 32)
-	ec.mu.Lock()
-	ec.subs = append(ec.subs, ch)
-	ec.mu.Unlock()
-	return ch, func() {
-		ec.mu.Lock()
-		for i, s := range ec.subs {
-			if s == ch {
-				ec.subs = append(ec.subs[:i], ec.subs[i+1:]...)
-				break
-			}
-		}
-		ec.mu.Unlock()
-	}
-}
-
-func (ec *eventCollector) waitFor(t *testing.T, typ string, timeout time.Duration) events.Event {
-	t.Helper()
-	ch, unsub := ec.subscribe()
-	defer unsub()
-	deadline := time.After(timeout)
-	for {
-		select {
-		case ev := <-ch:
-			if ev.Type == typ {
-				return ev
-			}
-		case <-deadline:
-			t.Fatalf("timed out waiting for event %q", typ)
-			return events.Event{}
-		}
-	}
-}
-
-func (ec *eventCollector) assertNone(t *testing.T, typ string, window time.Duration) {
-	t.Helper()
-	ch, unsub := ec.subscribe()
-	defer unsub()
-	deadline := time.After(window)
-	for {
-		select {
-		case ev := <-ch:
-			if ev.Type == typ {
-				t.Fatalf("unexpected event %q received", typ)
-			}
-		case <-deadline:
-			return
-		}
-	}
-}
-
 // newTestComputed creates a computedState with an eventCollector for testing.
 func newTestComputed(t *testing.T) (*computedState, *eventCollector) {
 	t.Helper()
diff --git a/server/lib/cdpmonitor/domains.go b/server/lib/cdpmonitor/domains.go
index 3c613edc..aef74f63 100644
--- a/server/lib/cdpmonitor/domains.go
+++ b/server/lib/cdpmonitor/domains.go
@@ -1,6 +1,9 @@
 package cdpmonitor
 
-import _ "embed"
+import (
+	"context"
+	_ "embed"
+)
 
 // bindingName is the JS function exposed via Runtime.addBinding.
 // Page JS calls this to fire Runtime.bindingCalled CDP events.
@@ -13,8 +16,51 @@ func isPageLikeTarget(targetType string) bool {
 	return targetType == "page" || targetType == "iframe"
 }
 
+// enableDomains enables CDP domains, registers the event binding, and starts
+// layout-shift observation. Failures are non-fatal.
+// Page-level domains (Page.enable, PerformanceTimeline.enable, Runtime.addBinding)
+// are skipped for worker and service_worker targets that don't support them.
+func (m *Monitor) enableDomains(ctx context.Context, sessionID string, targetType string) {
+	for _, method := range []string{
+		"Runtime.enable",
+		"Network.enable",
+	} {
+		if _, err := m.send(ctx, method, nil, sessionID); err != nil && ctx.Err() == nil {
+			m.log.Warn("cdpmonitor: failed to enable CDP domain", "method", method, "session", sessionID, "err", err)
+		}
+	}
+
+	if !isPageLikeTarget(targetType) {
+		return
+	}
+
+	if _, err := m.send(ctx, "Page.enable", nil, sessionID); err != nil && ctx.Err() == nil {
+		m.log.Warn("cdpmonitor: failed to enable CDP domain", "method", "Page.enable", "session", sessionID, "err", err)
+	}
+
+	if _, err := m.send(ctx, "Runtime.addBinding", map[string]any{
+		"name": bindingName,
+	}, sessionID); err != nil && ctx.Err() == nil {
+		m.log.Warn("cdpmonitor: failed to register JS binding", "session", sessionID, "err", err)
+	}
+
+	if _, err := m.send(ctx, "PerformanceTimeline.enable", map[string]any{
+		"eventTypes": []string{timelineEventLayoutShift},
+	}, sessionID); err != nil && ctx.Err() == nil {
+		m.log.Warn("cdpmonitor: failed to enable PerformanceTimeline", "session", sessionID, "err", err)
+	}
+}
+
 // injectedJS tracks clicks, keys, and scrolls via the __kernelEvent binding.
 // Layout shifts are handled natively by PerformanceTimeline.enable.
 //
 //go:embed interaction.js
 var injectedJS string
+
+// injectScript registers the interaction tracking JS for the given session.
+func (m *Monitor) injectScript(ctx context.Context, sessionID string) error {
+	_, err := m.send(ctx, "Page.addScriptToEvaluateOnNewDocument", map[string]any{
+		"source": injectedJS,
+	}, sessionID)
+	return err
+}
diff --git a/server/lib/cdpmonitor/handlers.go b/server/lib/cdpmonitor/handlers.go
new file mode 100644
index 00000000..83fe6ea9
--- /dev/null
+++ b/server/lib/cdpmonitor/handlers.go
@@ -0,0 +1,486 @@
+package cdpmonitor
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+)
+
+// logUnmarshalErr logs a Debug message when a handler can't parse CDP params.
+// These indicate Chrome sent an unexpected params shape, rare and non-actionable
+// at Warn/Error level, but useful in verbose mode.
+func (m *Monitor) logUnmarshalErr(method string, err error) {
+	m.log.Debug("cdpmonitor: failed to parse CDP params", "method", method, "err", err)
+}
+
+// publishEvent stamps common fields and publishes an event.
+func (m *Monitor) publishEvent(eventType string, category events.EventCategory, source events.Source, sourceEvent string, data json.RawMessage, sessionID string) {
+	src := source
+	src.Event = sourceEvent
+	if sessionID != "" {
+		if src.Metadata == nil {
+			src.Metadata = make(map[string]string)
+		}
+		src.Metadata[MetadataKeyCDPSessionID] = sessionID
+	}
+	m.publish(events.Event{
+		Ts:       time.Now().UnixMicro(),
+		Type:     eventType,
+		Category: category,
+		Source:   src,
+		Data:     data,
+	})
+}
+
+// dispatchEvent routes a CDP event to its handler.
+func (m *Monitor) dispatchEvent(msg cdpMessage) {
+	switch msg.Method {
+	case "Runtime.consoleAPICalled":
+		m.handleConsole(msg.Params, msg.SessionID)
+	case "Runtime.exceptionThrown":
+		m.handleExceptionThrown(msg.Params, msg.SessionID)
+	case "Runtime.bindingCalled":
+		m.handleBindingCalled(msg.Params, msg.SessionID)
+	case "Network.requestWillBeSent":
+		m.handleNetworkRequest(msg.Params, msg.SessionID)
+	case "Network.responseReceived":
+		m.handleResponseReceived(msg.Params, msg.SessionID)
+	case "Network.loadingFinished":
+		m.handleLoadingFinished(msg.Params, msg.SessionID)
+	case "Network.loadingFailed":
+		m.handleLoadingFailed(msg.Params, msg.SessionID)
+	case "Page.frameNavigated":
+		m.handleFrameNavigated(msg.Params, msg.SessionID)
+	case "Page.domContentEventFired":
+		m.handleDOMContentLoaded(msg.Params, msg.SessionID)
+	case "Page.loadEventFired":
+		m.handleLoadEventFired(msg.Params, msg.SessionID)
+	case "PerformanceTimeline.timelineEventAdded":
+		m.handleTimelineEvent(msg.Params, msg.SessionID)
+	case "Target.attachedToTarget":
+		m.handleAttachedToTarget(msg)
+	case "Target.targetCreated":
+		m.handleTargetCreated(msg.Params, msg.SessionID)
+	case "Target.targetDestroyed":
+		m.handleTargetDestroyed(msg.Params, msg.SessionID)
+	case "Target.detachedFromTarget":
+		m.handleDetachedFromTarget(msg.Params)
+	}
+}
+
+func (m *Monitor) handleConsole(params json.RawMessage, sessionID string) {
+	var p cdpConsoleParams
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Runtime.consoleAPICalled", err)
+		return
+	}
+
+	text := ""
+	if len(p.Args) > 0 {
+		text = consoleArgString(p.Args[0])
+	}
+	argValues := make([]string, 0, len(p.Args))
+	for _, a := range p.Args {
+		argValues = append(argValues, consoleArgString(a))
+	}
+	data, _ := json.Marshal(map[string]any{
+		"level":       p.Type,
+		"text":        text,
+		"args":        argValues,
+		"stack_trace": p.StackTrace,
+	})
+	m.publishEvent(EventConsoleLog, events.CategoryConsole, events.Source{Kind: events.KindCDP}, "Runtime.consoleAPICalled", data, sessionID)
+}
+
+func (m *Monitor) handleExceptionThrown(params json.RawMessage, sessionID string) {
+	var p cdpExceptionDetails
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Runtime.exceptionThrown", err)
+		return
+	}
+	data, _ := json.Marshal(map[string]any{
+		"text":        p.ExceptionDetails.Text,
+		"line":        p.ExceptionDetails.LineNumber,
+		"column":      p.ExceptionDetails.ColumnNumber,
+		"url":         p.ExceptionDetails.URL,
+		"stack_trace": p.ExceptionDetails.StackTrace,
+	})
+	m.publishEvent(EventConsoleError, events.CategoryConsole, events.Source{Kind: events.KindCDP}, "Runtime.exceptionThrown", data, sessionID)
+	m.tryScreenshot(m.getLifecycleCtx())
+}
+
+// bindingMinInterval is the minimum time between accepted __kernelEvent binding
+// calls per session. This caps throughput at 20 events/s per session, preventing
+// a misbehaving page from flooding the event pipeline.
+const bindingMinInterval = 50 * time.Millisecond
+
+// handleBindingCalled processes __kernelEvent binding calls from the page.
+func (m *Monitor) handleBindingCalled(params json.RawMessage, sessionID string) {
+	var p struct {
+		Name    string `json:"name"`
+		Payload string `json:"payload"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Runtime.bindingCalled", err)
+		return
+	}
+	if p.Name != bindingName {
+		return
+	}
+
+	payload := json.RawMessage(p.Payload)
+	if !json.Valid(payload) {
+		return
+	}
+	var header struct {
+		Type string `json:"type"`
+	}
+	if err := json.Unmarshal(payload, &header); err != nil {
+		return
+	}
+	switch header.Type {
+	case EventInteractionClick, EventInteractionKey, EventScrollSettled:
+	default:
+		return
+	}
+
+	// Rate-limit per (session, event type): cap at 20 events/s per pair so a
+	// misbehaving page cannot flood the event pipeline with a single event type.
+	now := time.Now()
+	rateKey := sessionID + ":" + header.Type
+	m.bindingRateMu.Lock()
+	last := m.bindingLastSeen[rateKey]
+	if now.Sub(last) < bindingMinInterval {
+		m.bindingRateMu.Unlock()
+		return
+	}
+	m.bindingLastSeen[rateKey] = now
+	m.bindingRateMu.Unlock()
+
+	m.publishEvent(header.Type, events.CategoryInteraction, events.Source{Kind: events.KindCDP}, "Runtime.bindingCalled", payload, sessionID)
+}
+
+// handleTimelineEvent processes PerformanceTimeline layout-shift events.
+func (m *Monitor) handleTimelineEvent(params json.RawMessage, sessionID string) {
+	var p struct {
+		Event struct {
+			Type        string          `json:"type"`
+			LayoutShift json.RawMessage `json:"layoutShiftDetails,omitempty"`
+		} `json:"event"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("PerformanceTimeline.timelineEventAdded", err)
+		return
+	}
+	if p.Event.Type != timelineEventLayoutShift {
+		return
+	}
+	m.publishEvent(EventLayoutShift, events.CategoryPage, events.Source{Kind: events.KindCDP}, "PerformanceTimeline.timelineEventAdded", params, sessionID)
+	m.computed.onLayoutShift()
+}
+
+// handleNetworkRequest publishes network_request events.
+// NOTE: events include raw headers, post_data, and (on response) truncated
+// bodies which may contain cookies, bearer tokens, or other credentials.
+// This mirrors what CDP/DevTools itself exposes. Consumers should treat the
+// event stream as privileged data; opt-in redaction can be added later.
+func (m *Monitor) handleNetworkRequest(params json.RawMessage, sessionID string) {
+	var p cdpNetworkRequestParams
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Network.requestWillBeSent", err)
+		return
+	}
+	// Extract only the initiator type; the stack trace is too verbose and dominates event size.
+	var initiatorType string
+	var raw struct {
+		Type string `json:"type"`
+	}
+	if json.Unmarshal(p.Initiator, &raw) == nil {
+		initiatorType = raw.Type
+	}
+
+	// Redirects reuse the same requestId and fire additional requestWillBeSent
+	// events, but only a single loadingFinished fires per chain. Only increment
+	// netPending for genuinely new requests to avoid permanently inflating the
+	// counter and blocking network_idle.
+	m.pendReqMu.Lock()
+	existing, isRedirect := m.pendingRequests[p.RequestID]
+	addedAt := existing.addedAt
+	if !isRedirect {
+		addedAt = time.Now()
+	}
+	m.pendingRequests[p.RequestID] = networkReqState{
+		sessionID:    sessionID,
+		method:       p.Request.Method,
+		url:          p.Request.URL,
+		headers:      p.Request.Headers,
+		postData:     p.Request.PostData,
+		resourceType: p.ResourceType,
+		addedAt:      addedAt,
+	}
+	m.pendReqMu.Unlock()
+	ev := map[string]any{
+		"method":         p.Request.Method,
+		"url":            p.Request.URL,
+		"headers":        p.Request.Headers,
+		"initiator_type": initiatorType,
+	}
+	if p.Request.PostData != "" {
+		ev["post_data"] = p.Request.PostData
+	}
+	if p.ResourceType != "" {
+		ev["resource_type"] = p.ResourceType
+	}
+	data, _ := json.Marshal(ev)
+	m.publishEvent(EventNetworkRequest, events.CategoryNetwork, events.Source{Kind: events.KindCDP}, "Network.requestWillBeSent", data, sessionID)
+	if !isRedirect {
+		m.computed.onRequest()
+	}
+}
+
+func (m *Monitor) handleResponseReceived(params json.RawMessage, _ string) {
+	var p cdpResponseReceivedParams
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Network.responseReceived", err)
+		return
+	}
+	m.pendReqMu.Lock()
+	if state, ok := m.pendingRequests[p.RequestID]; ok {
+		state.status = p.Response.Status
+		state.statusText = p.Response.StatusText
+		state.resHeaders = p.Response.Headers
+		state.mimeType = p.Response.MimeType
+		m.pendingRequests[p.RequestID] = state
+	}
+	m.pendReqMu.Unlock()
+}
+
+func (m *Monitor) handleLoadingFinished(params json.RawMessage, sessionID string) {
+	var p struct {
+		RequestID string `json:"requestId"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Network.loadingFinished", err)
+		return
+	}
+	m.pendReqMu.Lock()
+	state, ok := m.pendingRequests[p.RequestID]
+	if ok {
+		delete(m.pendingRequests, p.RequestID)
+	}
+	m.pendReqMu.Unlock()
+	if !ok {
+		return
+	}
+	m.computed.onLoadingFinished()
+	// Fetch response body async to avoid blocking readLoop; binary types are skipped.
+	m.asyncWg.Go(func() {
+		body := m.fetchResponseBody(p.RequestID, sessionID, state)
+		ev := map[string]any{
+			"method":  state.method,
+			"url":     state.url,
+			"status":  state.status,
+			"headers": state.resHeaders,
+		}
+		if state.statusText != "" {
+			ev["status_text"] = state.statusText
+		}
+		if state.mimeType != "" {
+			ev["mime_type"] = state.mimeType
+		}
+		if state.resourceType != "" {
+			ev["resource_type"] = state.resourceType
+		}
+		if body != "" {
+			ev["body"] = body
+		}
+		data, _ := json.Marshal(ev)
+		m.publishEvent(EventNetworkResponse, events.CategoryNetwork, events.Source{Kind: events.KindCDP}, "Network.loadingFinished", data, sessionID)
+	})
+}
+
+// fetchResponseBody retrieves and truncates the response body for textual resources.
+func (m *Monitor) fetchResponseBody(requestID, sessionID string, state networkReqState) string {
+	if !isTextualResource(state.resourceType, state.mimeType) {
+		return ""
+	}
+	result, err := m.send(m.getLifecycleCtx(), "Network.getResponseBody", map[string]any{
+		"requestId": requestID,
+	}, sessionID)
+	if err != nil {
+		return ""
+	}
+	var resp struct {
+		Body          string `json:"body"`
+		Base64Encoded bool   `json:"base64Encoded"`
+	}
+	if json.Unmarshal(result, &resp) != nil {
+		return ""
+	}
+	body := resp.Body
+	if resp.Base64Encoded {
+		decoded, err := base64.StdEncoding.DecodeString(body)
+		if err != nil {
+			return ""
+		}
+		body = string(decoded)
+	}
+	return truncateBody(body, bodyCapFor(state.mimeType))
+}
+
+func (m *Monitor) handleLoadingFailed(params json.RawMessage, sessionID string) {
+	var p struct {
+		RequestID string `json:"requestId"`
+		ErrorText string `json:"errorText"`
+		Canceled  bool   `json:"canceled"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Network.loadingFailed", err)
+		return
+	}
+	m.pendReqMu.Lock()
+	state, ok := m.pendingRequests[p.RequestID]
+	if ok {
+		delete(m.pendingRequests, p.RequestID)
+	}
+	m.pendReqMu.Unlock()
+
+	ev := map[string]any{
+		"error_text": p.ErrorText,
+		"canceled":   p.Canceled,
+	}
+	if ok {
+		ev["url"] = state.url
+	}
+	data, _ := json.Marshal(ev)
+	m.publishEvent(EventNetworkLoadingFailed, events.CategoryNetwork, events.Source{Kind: events.KindCDP}, "Network.loadingFailed", data, sessionID)
+	if ok {
+		m.computed.onLoadingFinished()
+	}
+}
+
+func (m *Monitor) handleFrameNavigated(params json.RawMessage, sessionID string) {
+	var p struct {
+		Frame struct {
+			ID       string `json:"id"`
+			ParentID string `json:"parentId"`
+			URL      string `json:"url"`
+		} `json:"frame"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Page.frameNavigated", err)
+		return
+	}
+	data, _ := json.Marshal(map[string]any{
+		"url":             p.Frame.URL,
+		"frame_id":        p.Frame.ID,
+		"parent_frame_id": p.Frame.ParentID,
+	})
+	m.publishEvent(EventNavigation, events.CategoryPage, events.Source{Kind: events.KindCDP}, "Page.frameNavigated", data, sessionID)
+
+	// Only reset state for top-level navigations; subframe (iframe) navigations
+	// should not disrupt main-page tracking.
+	if p.Frame.ParentID == "" {
+		m.mainSessionID.Store(sessionID)
+		m.pendReqMu.Lock()
+		for id, req := range m.pendingRequests {
+			if req.sessionID == sessionID {
+				delete(m.pendingRequests, id)
+			}
+		}
+		inflight := len(m.pendingRequests)
+		// Reset computed state while still holding pendReqMu so new requests
+		// arriving concurrently can't increment netPending before the reset.
+		m.computed.resetOnNavigation(inflight)
+		m.pendReqMu.Unlock()
+	}
+}
+
+func (m *Monitor) handleDOMContentLoaded(params json.RawMessage, sessionID string) {
+	m.publishEvent(EventDOMContentLoaded, events.CategoryPage, events.Source{Kind: events.KindCDP}, "Page.domContentEventFired", params, sessionID)
+	// Only advance the state machine for the main frame; subframe events arrive
+	// on their own sessionId and would trigger navigation_settled prematurely.
+	if m.mainSessionID.Load() == sessionID {
+		m.computed.onDOMContentLoaded()
+	}
+}
+
+func (m *Monitor) handleLoadEventFired(params json.RawMessage, sessionID string) {
+	m.publishEvent(EventPageLoad, events.CategoryPage, events.Source{Kind: events.KindCDP}, "Page.loadEventFired", params, sessionID)
+	if m.mainSessionID.Load() == sessionID {
+		m.computed.onPageLoad()
+		m.tryScreenshot(m.getLifecycleCtx())
+	}
+}
+
+// handleAttachedToTarget stores the new session then enables domains and injects script.
+func (m *Monitor) handleAttachedToTarget(msg cdpMessage) {
+	var params cdpAttachedToTargetParams
+	if err := json.Unmarshal(msg.Params, &params); err != nil {
+		m.logUnmarshalErr("Target.attachedToTarget", err)
+		return
+	}
+	m.sessionsMu.Lock()
+	m.sessions[params.SessionID] = targetInfo{
+		targetID:   params.TargetInfo.TargetID,
+		url:        params.TargetInfo.URL,
+		targetType: params.TargetInfo.Type,
+	}
+	m.sessionsMu.Unlock()
+
+	targetType := params.TargetInfo.Type
+	// Async to avoid blocking the readLoop.
+	m.asyncWg.Go(func() {
+		ctx := m.getLifecycleCtx()
+		m.enableDomains(ctx, params.SessionID, targetType)
+		if isPageLikeTarget(targetType) {
+			_ = m.injectScript(ctx, params.SessionID)
+		}
+	})
+}
+
+func (m *Monitor) handleTargetCreated(params json.RawMessage, sessionID string) {
+	var p cdpTargetCreatedParams
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Target.targetCreated", err)
+		return
+	}
+	data, _ := json.Marshal(map[string]any{
+		"target_id":   p.TargetInfo.TargetID,
+		"target_type": p.TargetInfo.Type,
+		"url":         p.TargetInfo.URL,
+	})
+	m.publishEvent(EventTargetCreated, events.CategoryPage, events.Source{Kind: events.KindCDP}, "Target.targetCreated", data, sessionID)
+}
+
+func (m *Monitor) handleTargetDestroyed(params json.RawMessage, sessionID string) {
+	var p struct {
+		TargetID string `json:"targetId"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Target.targetDestroyed", err)
+		return
+	}
+	data, _ := json.Marshal(map[string]any{
+		"target_id": p.TargetID,
+	})
+	m.publishEvent(EventTargetDestroyed, events.CategoryPage, events.Source{Kind: events.KindCDP}, "Target.targetDestroyed", data, sessionID)
+}
+
+func (m *Monitor) handleDetachedFromTarget(params json.RawMessage) {
+	var p struct {
+		SessionID string `json:"sessionId"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		m.logUnmarshalErr("Target.detachedFromTarget", err)
+		return
+	}
+	if p.SessionID == "" {
+		return
+	}
+	m.sessionsMu.Lock()
+	delete(m.sessions, p.SessionID)
+	m.sessionsMu.Unlock()
+}
diff --git a/server/lib/cdpmonitor/handlers_test.go b/server/lib/cdpmonitor/handlers_test.go
new file mode 100644
index 00000000..1518b412
--- /dev/null
+++ b/server/lib/cdpmonitor/handlers_test.go
@@ -0,0 +1,358 @@
+package cdpmonitor
+
+import (
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConsoleEvents(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	_, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+
+	t.Run("console_log", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.consoleAPICalled",
+			"params": map[string]any{
+				"type": "log",
+				"args": []any{map[string]any{"type": "string", "value": "hello world"}},
+			},
+		})
+		ev := ec.waitFor(t, "console_log", 2*time.Second)
+		assert.Equal(t, events.CategoryConsole, ev.Category)
+		assert.Equal(t, events.KindCDP, ev.Source.Kind)
+		assert.Equal(t, "Runtime.consoleAPICalled", ev.Source.Event)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "log", data["level"])
+		assert.Equal(t, "hello world", data["text"])
+	})
+
+	t.Run("exception_thrown", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.exceptionThrown",
+			"params": map[string]any{
+				"timestamp": 1234.5,
+				"exceptionDetails": map[string]any{
+					"text":         "Uncaught TypeError",
+					"lineNumber":   42,
+					"columnNumber": 7,
+					"url":          "https://example.com/app.js",
+				},
+			},
+		})
+		ev := ec.waitFor(t, "console_error", 2*time.Second)
+		assert.Equal(t, events.CategoryConsole, ev.Category)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "Uncaught TypeError", data["text"])
+		assert.Equal(t, float64(42), data["line"])
+	})
+
+	t.Run("non_string_args", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.consoleAPICalled",
+			"params": map[string]any{
+				"type": "log",
+				"args": []any{
+					map[string]any{"type": "number", "value": 42},
+					map[string]any{"type": "object", "value": map[string]any{"key": "val"}},
+					map[string]any{"type": "undefined"},
+				},
+			},
+		})
+		ev := ec.waitForNew(t, "console_log", 2*time.Second)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		args := data["args"].([]any)
+		assert.Equal(t, "42", args[0])
+		assert.Contains(t, args[1], "key")
+		assert.Equal(t, "undefined", args[2])
+	})
+}
+
+func TestNetworkEvents(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	var getBodyCalled atomic.Bool
+	responder := func(msg cdpMessage) any {
+		if msg.Method == "Network.getResponseBody" {
+			getBodyCalled.Store(true)
+			return map[string]any{
+				"id":     msg.ID,
+				"result": map[string]any{"body": `{"ok":true}`, "base64Encoded": false},
+			}
+		}
+		return nil
+	}
+	_, ec, cleanup := startMonitor(t, srv, responder)
+	defer cleanup()
+
+	t.Run("request_and_response", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.requestWillBeSent",
+			"params": map[string]any{
+				"requestId":    "req-001",
+				"resourceType": "XHR",
+				"request": map[string]any{
+					"method":  "POST",
+					"url":     "https://api.example.com/data",
+					"headers": map[string]any{"Content-Type": "application/json"},
+				},
+				"initiator": map[string]any{"type": "script"},
+			},
+		})
+		ev := ec.waitFor(t, "network_request", 2*time.Second)
+		assert.Equal(t, events.CategoryNetwork, ev.Category)
+		assert.Equal(t, "Network.requestWillBeSent", ev.Source.Event)
+
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "POST", data["method"])
+		assert.Equal(t, "https://api.example.com/data", data["url"])
+
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.responseReceived",
+			"params": map[string]any{
+				"requestId": "req-001",
+				"response": map[string]any{
+					"status": 200, "statusText": "OK",
+					"headers": map[string]any{"Content-Type": "application/json"}, "mimeType": "application/json",
+				},
+			},
+		})
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.loadingFinished",
+			"params": map[string]any{"requestId": "req-001"},
+		})
+
+		ev2 := ec.waitFor(t, "network_response", 3*time.Second)
+		assert.Equal(t, "Network.loadingFinished", ev2.Source.Event)
+		var data2 map[string]any
+		require.NoError(t, json.Unmarshal(ev2.Data, &data2))
+		assert.Equal(t, float64(200), data2["status"])
+		assert.NotEmpty(t, data2["body"])
+	})
+
+	t.Run("loading_failed", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.requestWillBeSent",
+			"params": map[string]any{
+				"requestId": "req-002",
+				"request":   map[string]any{"method": "GET", "url": "https://fail.example.com/"},
+			},
+		})
+		ec.waitForNew(t, "network_request", 2*time.Second)
+
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.loadingFailed",
+			"params": map[string]any{
+				"requestId": "req-002",
+				"errorText": "net::ERR_CONNECTION_REFUSED",
+				"canceled":  false,
+			},
+		})
+		ev := ec.waitFor(t, "network_loading_failed", 2*time.Second)
+		assert.Equal(t, events.CategoryNetwork, ev.Category)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "net::ERR_CONNECTION_REFUSED", data["error_text"])
+	})
+
+	t.Run("binary_resource_skips_body", func(t *testing.T) {
+		getBodyCalled.Store(false)
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.requestWillBeSent",
+			"params": map[string]any{
+				"requestId":    "img-001",
+				"resourceType": "Image",
+				"request":      map[string]any{"method": "GET", "url": "https://example.com/photo.png"},
+			},
+		})
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.responseReceived",
+			"params": map[string]any{
+				"requestId": "img-001",
+				"response":  map[string]any{"status": 200, "statusText": "OK", "headers": map[string]any{}, "mimeType": "image/png"},
+			},
+		})
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Network.loadingFinished",
+			"params": map[string]any{"requestId": "img-001"},
+		})
+
+		ev := ec.waitForNew(t, "network_response", 3*time.Second)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Nil(t, data["body"], "binary resource should not have body field")
+		assert.False(t, getBodyCalled.Load(), "should not call getResponseBody for images")
+	})
+}
+
+func TestPageEvents(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	_, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Page.frameNavigated",
+		"params": map[string]any{
+			"frame": map[string]any{"id": "frame-1", "url": "https://example.com/page"},
+		},
+	})
+	ev := ec.waitFor(t, "navigation", 2*time.Second)
+	assert.Equal(t, events.CategoryPage, ev.Category)
+	assert.Equal(t, "Page.frameNavigated", ev.Source.Event)
+	var data map[string]any
+	require.NoError(t, json.Unmarshal(ev.Data, &data))
+	assert.Equal(t, "https://example.com/page", data["url"])
+
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Page.domContentEventFired",
+		"params": map[string]any{"timestamp": 1000.0},
+	})
+	ev2 := ec.waitFor(t, "dom_content_loaded", 2*time.Second)
+	assert.Equal(t, events.CategoryPage, ev2.Category)
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Page.loadEventFired",
+		"params": map[string]any{"timestamp": 1001.0},
+	})
+	ev3 := ec.waitFor(t, "page_load", 2*time.Second)
+	assert.Equal(t, events.CategoryPage, ev3.Category)
+}
+
+func TestTargetEvents(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	_, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Target.targetCreated",
+		"params": map[string]any{
+			"targetInfo": map[string]any{"targetId": "t-1", "type": "page", "url": "https://new.example.com"},
+		},
+	})
+	ev := ec.waitFor(t, "target_created", 2*time.Second)
+	assert.Equal(t, events.CategoryPage, ev.Category)
+	var data map[string]any
+	require.NoError(t, json.Unmarshal(ev.Data, &data))
+	assert.Equal(t, "t-1", data["target_id"])
+
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Target.targetDestroyed",
+		"params": map[string]any{"targetId": "t-1"},
+	})
+	ev2 := ec.waitFor(t, "target_destroyed", 2*time.Second)
+	assert.Equal(t, events.CategoryPage, ev2.Category)
+}
+
+func TestBindingAndTimeline(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	_, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+
+	t.Run("interaction_click", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.bindingCalled",
+			"params": map[string]any{
+				"name":    "__kernelEvent",
+				"payload": `{"type":"interaction_click","x":10,"y":20,"selector":"button","tag":"BUTTON","text":"OK"}`,
+			},
+		})
+		ev := ec.waitFor(t, "interaction_click", 2*time.Second)
+		assert.Equal(t, events.CategoryInteraction, ev.Category)
+		assert.Equal(t, "Runtime.bindingCalled", ev.Source.Event)
+	})
+
+	t.Run("scroll_settled", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.bindingCalled",
+			"params": map[string]any{
+				"name":    "__kernelEvent",
+				"payload": `{"type":"scroll_settled","from_x":0,"from_y":0,"to_x":0,"to_y":500,"target_selector":"body"}`,
+			},
+		})
+		ev := ec.waitFor(t, "scroll_settled", 2*time.Second)
+		assert.Equal(t, events.CategoryInteraction, ev.Category)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, float64(500), data["to_y"])
+	})
+
+	t.Run("layout_shift", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "PerformanceTimeline.timelineEventAdded",
+			"params": map[string]any{
+				"event": map[string]any{"type": "layout-shift"},
+			},
+		})
+		ev := ec.waitFor(t, "layout_shift", 2*time.Second)
+		assert.Equal(t, events.KindCDP, ev.Source.Kind)
+		assert.Equal(t, "PerformanceTimeline.timelineEventAdded", ev.Source.Event)
+	})
+
+	t.Run("unknown_binding_ignored", func(t *testing.T) {
+		srv.sendToMonitor(t, map[string]any{
+			"method": "Runtime.bindingCalled",
+			"params": map[string]any{
+				"name":    "someOtherBinding",
+				"payload": `{"type":"interaction_click"}`,
+			},
+		})
+		ec.assertNone(t, "interaction_click", 100*time.Millisecond)
+	})
+
+	t.Run("rate_limited_per_session", func(t *testing.T) {
+		// Send two binding events back-to-back within the 50ms window.
+		// Only the first should produce a published event.
+		before := func() int {
+			ec.mu.Lock()
+			defer ec.mu.Unlock()
+			count := 0
+			for _, ev := range ec.events {
+				if ev.Type == EventInteractionClick {
+					count++
+				}
+			}
+			return count
+		}
+		countBefore := before()
+
+		for range 3 {
+			srv.sendToMonitor(t, map[string]any{
+				"method": "Runtime.bindingCalled",
+				"params": map[string]any{
+					"name":    "__kernelEvent",
+					"payload": `{"type":"interaction_click","x":1,"y":1,"selector":"a","tag":"A","text":"x"}`,
+				},
+			})
+		}
+
+		// Wait a bit for async delivery, then check only 1 new event was published.
+		time.Sleep(200 * time.Millisecond)
+		ec.mu.Lock()
+		countAfter := 0
+		for _, ev := range ec.events {
+			if ev.Type == EventInteractionClick {
+				countAfter++
+			}
+		}
+		ec.mu.Unlock()
+		assert.Equal(t, countBefore+1, countAfter, "rate limiter should have dropped the 2nd and 3rd events")
+	})
+}
diff --git a/server/lib/cdpmonitor/monitor.go b/server/lib/cdpmonitor/monitor.go
index e7cf747f..a0323f8f 100644
--- a/server/lib/cdpmonitor/monitor.go
+++ b/server/lib/cdpmonitor/monitor.go
@@ -2,8 +2,14 @@ package cdpmonitor
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"sync"
 	"sync/atomic"
+	"time"
 
+	"github.com/coder/websocket"
 	"github.com/kernel/kernel-images/server/lib/events"
 )
 
@@ -16,15 +22,83 @@ type UpstreamProvider interface {
 // PublishFunc publishes an Event to the pipeline.
 type PublishFunc func(ev events.Event)
 
+const wsReadLimit = 8 * 1024 * 1024
+
 // Monitor manages a CDP WebSocket connection with auto-attach session fan-out.
-// Single-use per capture session: call Start to begin, Stop to tear down.
+//
+// Lock ordering (outer → inner):
+//
+//	restartMu → lifeMu → pendReqMu → computed.mu → pendMu → sessionsMu
+//
+// Never acquire a lock that appears later in this order while holding an
+// earlier one, to prevent deadlock.
+//
+// WebSocket concurrency: coder/websocket guarantees that one concurrent Read
+// and one concurrent Write are safe. The readLoop holds the sole Read; all
+// writes go through send, which serialises them with conn.Write's internal
+// lock. No external write mutex is needed.
 type Monitor struct {
+	upstreamMgr UpstreamProvider
+	publish     PublishFunc
+	displayNum  int
+	log         *slog.Logger
+
+	// lifeMu serializes Start, Stop, and restartReadLoop to prevent races on
+	// conn, lifecycleCtx, cancel, and done.
+	lifeMu sync.Mutex
+	conn   *websocket.Conn
+
+	nextID  atomic.Int64
+	pendMu  sync.Mutex
+	pending map[int64]chan cdpMessage
+
+	sessionsMu    sync.RWMutex
+	sessions      map[string]targetInfo // sessionID → targetInfo
+	mainSessionID atomic.Value          // string; set on first top-level frameNavigated, cleared on reconnect
+
+	pendReqMu       sync.Mutex
+	pendingRequests map[string]networkReqState // requestId → networkReqState
+
+	computed *computedState
+
+	lastScreenshotAt   atomic.Int64                                              // unix millis of last capture
+	screenshotInFlight atomic.Bool                                               // true while a captureScreenshot goroutine is running
+	screenshotFn       func(ctx context.Context, displayNum int) ([]byte, error) // nil → real ffmpeg
+
+	// bindingRateMu guards bindingLastSeen.
+	bindingRateMu   sync.Mutex
+	bindingLastSeen map[string]time.Time // sessionID → last accepted binding event time
+
+	// asyncWg tracks all goroutines except readLoop (which is tracked via done).
+	// subscribeToUpstream and sweepPendingRequests are included so Stop() can
+	// wait for them to exit before returning.
+	asyncWg   sync.WaitGroup
+	restartMu sync.Mutex // serializes handleUpstreamRestart to prevent overlapping reconnects
+
+	lifecycleCtx context.Context // cancelled on Stop()
+	cancel       context.CancelFunc
+	done         chan struct{}
+	readReady    chan struct{} // closed when readLoop has started reading
+
 	running atomic.Bool
 }
 
 // New creates a Monitor. displayNum is the X display for ffmpeg screenshots.
-func New(_ UpstreamProvider, _ PublishFunc, _ int) *Monitor {
-	return &Monitor{}
+func New(upstreamMgr UpstreamProvider, publish PublishFunc, displayNum int, log *slog.Logger) *Monitor {
+	m := &Monitor{
+		upstreamMgr:     upstreamMgr,
+		publish:         publish,
+		displayNum:      displayNum,
+		log:             log,
+		sessions:        make(map[string]targetInfo),
+		pending:         make(map[int64]chan cdpMessage),
+		pendingRequests: make(map[string]networkReqState),
+		bindingLastSeen: make(map[string]time.Time),
+	}
+	m.computed = newComputedState(publish)
+	m.lifecycleCtx = context.Background()
+	m.mainSessionID.Store(mainSessionUnset)
+	return m
 }
 
 // IsRunning reports whether the monitor is actively capturing.
@@ -32,10 +106,503 @@ func (m *Monitor) IsRunning() bool {
 	return m.running.Load()
 }
 
+// Health returns a point-in-time snapshot of internal counters useful for
+// debugging and operational visibility.
+func (m *Monitor) Health() MonitorHealth {
+	m.pendMu.Lock()
+	pendCmds := len(m.pending)
+	m.pendMu.Unlock()
+
+	m.pendReqMu.Lock()
+	pendReqs := len(m.pendingRequests)
+	m.pendReqMu.Unlock()
+
+	m.sessionsMu.RLock()
+	sessions := len(m.sessions)
+	m.sessionsMu.RUnlock()
+
+	return MonitorHealth{
+		Running:         m.running.Load(),
+		PendingCommands: pendCmds,
+		PendingRequests: pendReqs,
+		Sessions:        sessions,
+	}
+}
+
+// getLifecycleCtx returns the current lifecycle context under lifeMu.
+func (m *Monitor) getLifecycleCtx() context.Context {
+	m.lifeMu.Lock()
+	ctx := m.lifecycleCtx
+	m.lifeMu.Unlock()
+	return ctx
+}
+
 // Start begins CDP capture. Restarts if already running.
+// Not concurrency-safe; callers must serialize Start calls.
 func (m *Monitor) Start(_ context.Context) error {
+	m.Stop() // no-op if not running
+
+	devtoolsURL := m.upstreamMgr.Current()
+	if devtoolsURL == "" {
+		return fmt.Errorf("cdpmonitor: no DevTools URL available")
+	}
+
+	// Use background context so the monitor outlives the caller's request context.
+	ctx, cancel := context.WithCancel(context.Background())
+
+	conn, _, err := websocket.Dial(ctx, devtoolsURL, nil)
+	if err != nil {
+		cancel()
+		return fmt.Errorf("cdpmonitor: dial %s: %w", devtoolsURL, err)
+	}
+	conn.SetReadLimit(wsReadLimit)
+
+	m.lifeMu.Lock()
+	m.conn = conn
+	m.lifecycleCtx = ctx
+	m.cancel = cancel
+	m.done = make(chan struct{})
+	m.readReady = make(chan struct{})
+	m.lifeMu.Unlock()
+
+	m.running.Store(true)
+	m.log.Info("cdpmonitor: started", "url", devtoolsURL)
+
+	go m.readLoop(ctx)
+	m.asyncWg.Go(func() { m.subscribeToUpstream(ctx) })
+	m.asyncWg.Go(func() { m.sweepPendingRequests(ctx) })
+	m.asyncWg.Go(func() { m.initSession(ctx) })
+
 	return nil
 }
 
-// Stop tears down the monitor. Safe to call multiple times.
-func (m *Monitor) Stop() {}
+// Stop cancels the context and waits for goroutines to exit.
+func (m *Monitor) Stop() {
+	if !m.running.Swap(false) {
+		return
+	}
+	m.log.Info("cdpmonitor: stopping")
+
+	m.lifeMu.Lock()
+	if m.cancel != nil {
+		m.cancel()
+	}
+	done := m.done
+	m.lifeMu.Unlock()
+
+	if done != nil {
+		<-done
+	}
+
+	// Wait for all in-flight async goroutines (fetchResponseBody, enableDomains,
+	// screenshots) to finish before closing the connection they may be writing to.
+	m.asyncWg.Wait()
+
+	m.lifeMu.Lock()
+	if m.conn != nil {
+		_ = m.conn.Close(websocket.StatusNormalClosure, "stopped")
+		m.conn = nil
+	}
+	m.lifeMu.Unlock()
+
+	m.clearState()
+	m.log.Info("cdpmonitor: stopped")
+}
+
+// clearState resets sessions, pending requests, and computed state.
+// It also fails all in-flight send() calls so their goroutines are unblocked.
+func (m *Monitor) clearState() {
+	m.sessionsMu.Lock()
+	m.sessions = make(map[string]targetInfo)
+	m.sessionsMu.Unlock()
+	m.mainSessionID.Store(mainSessionUnset)
+
+	m.pendReqMu.Lock()
+	m.pendingRequests = make(map[string]networkReqState)
+	m.pendReqMu.Unlock()
+
+	m.bindingRateMu.Lock()
+	m.bindingLastSeen = make(map[string]time.Time)
+	m.bindingRateMu.Unlock()
+
+	m.failPendingCommands()
+
+	// pendingRequests is already empty above, so inflight=0 is correct.
+	m.computed.resetOnNavigation(0)
+}
+
+const pendingRequestTTL = 5 * time.Minute
+const sweepInterval = 1 * time.Minute
+
+// sweepPendingRequests periodically evicts networkReqState entries that have
+// been in the map for longer than pendingRequestTTL. This bounds map growth on
+// long-lived SPAs where loadingFinished never arrives (e.g. tabs closed mid-flight).
+// It exits when ctx is cancelled (Stop/reconnect).
+func (m *Monitor) sweepPendingRequests(ctx context.Context) {
+	ticker := time.NewTicker(sweepInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case now := <-ticker.C:
+			m.pendReqMu.Lock()
+			for id, state := range m.pendingRequests {
+				if now.Sub(state.addedAt) > pendingRequestTTL {
+					delete(m.pendingRequests, id)
+					m.computed.onLoadingFinished() // keep netPending consistent
+				}
+			}
+			m.pendReqMu.Unlock()
+		}
+	}
+}
+
+// failPendingCommands unblocks all in-flight send() calls by delivering an
+// error response. This prevents goroutine leaks when the connection is torn
+// down during reconnect.
+func (m *Monitor) failPendingCommands() {
+	m.pendMu.Lock()
+	old := m.pending
+	m.pending = make(map[int64]chan cdpMessage)
+	m.pendMu.Unlock()
+
+	disconnectErr := &cdpError{Code: -1, Message: "connection closed"}
+	for _, ch := range old {
+		select {
+		case ch <- cdpMessage{Error: disconnectErr}:
+		default:
+		}
+	}
+}
+
+// readLoop reads CDP messages, routing responses to pending callers and dispatching events.
+// On read error (WS drop) this goroutine returns. Reconnection is driven by
+// subscribeToUpstream: the UpstreamProvider always pushes a fresh devtools URL
+// when the browser process restarts, so same-URL redial is not needed here.
+func (m *Monitor) readLoop(ctx context.Context) {
+	m.lifeMu.Lock()
+	done := m.done
+	conn := m.conn
+	readReady := m.readReady
+	m.lifeMu.Unlock()
+	defer close(done)
+
+	if conn == nil {
+		return
+	}
+
+	// Signal that readLoop is ready to receive responses.
+	close(readReady)
+
+	for {
+		_, b, err := conn.Read(ctx)
+		if err != nil {
+			if ctx.Err() == nil {
+				m.log.Warn("cdpmonitor: read loop exiting on unexpected error", "err", err)
+			}
+			return
+		}
+
+		var msg cdpMessage
+		if err := json.Unmarshal(b, &msg); err != nil {
+			m.log.Warn("cdpmonitor: dropping malformed CDP message", "err", err)
+			continue
+		}
+
+		if msg.ID != nil {
+			m.pendMu.Lock()
+			ch, ok := m.pending[*msg.ID]
+			m.pendMu.Unlock()
+			if ok {
+				select {
+				case ch <- msg:
+				default:
+					// send() already timed out and deregistered; discard.
+				}
+			}
+			continue
+		}
+
+		m.dispatchEvent(msg)
+	}
+}
+
+const sendTimeout = 30 * time.Second
+
+// send issues a CDP command and blocks until the response arrives.
+// A 30 s deadline is applied so a non-responsive Chrome cannot stall
+// callers indefinitely; the caller's own deadline (if shorter) wins.
+func (m *Monitor) send(ctx context.Context, method string, params any, sessionID string) (json.RawMessage, error) {
+	ctx, cancel := context.WithTimeout(ctx, sendTimeout)
+	defer cancel()
+
+	id := m.nextID.Add(1)
+
+	var rawParams json.RawMessage
+	if params != nil {
+		b, err := json.Marshal(params)
+		if err != nil {
+			return nil, fmt.Errorf("marshal params: %w", err)
+		}
+		rawParams = b
+	}
+
+	req := cdpMessage{ID: &id, Method: method, Params: rawParams, SessionID: sessionID}
+	reqBytes, err := json.Marshal(req)
+	if err != nil {
+		return nil, fmt.Errorf("marshal request: %w", err)
+	}
+
+	ch := make(chan cdpMessage, 1)
+	m.pendMu.Lock()
+	m.pending[id] = ch
+	m.pendMu.Unlock()
+	defer func() {
+		m.pendMu.Lock()
+		delete(m.pending, id)
+		m.pendMu.Unlock()
+	}()
+
+	m.lifeMu.Lock()
+	conn := m.conn
+	m.lifeMu.Unlock()
+	if conn == nil {
+		return nil, fmt.Errorf("cdpmonitor: connection not open")
+	}
+
+	// coder/websocket allows concurrent Read + Write on the same Conn.
+	if err := conn.Write(ctx, websocket.MessageText, reqBytes); err != nil {
+		return nil, fmt.Errorf("write: %w", err)
+	}
+
+	select {
+	case resp := <-ch:
+		if resp.Error != nil {
+			return nil, resp.Error
+		}
+		return resp.Result, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// initSession enables CDP domains, injects the interaction-tracking script,
+// and manually attaches to any targets already open when the monitor started.
+// It waits for readLoop to be ready before sending any commands.
+func (m *Monitor) initSession(ctx context.Context) {
+	m.lifeMu.Lock()
+	readReady := m.readReady
+	m.lifeMu.Unlock()
+	select {
+	case <-readReady:
+	case <-ctx.Done():
+		return
+	}
+
+	if _, err := m.send(ctx, "Target.setAutoAttach", map[string]any{
+		"autoAttach":             true,
+		"waitForDebuggerOnStart": false,
+		"flatten":                true,
+	}, ""); err != nil && ctx.Err() == nil {
+		// Without auto-attach the monitor will never see new targets: treat as fatal.
+		m.log.Error("cdpmonitor: Target.setAutoAttach failed — monitor will not observe new targets", "err", err)
+		m.publish(events.Event{
+			Ts:       time.Now().UnixMicro(),
+			Type:     EventMonitorInitFailed,
+			Category: events.CategorySystem,
+			Source:   events.Source{Kind: events.KindLocalProcess},
+			Data:     json.RawMessage(`{"step":"Target.setAutoAttach"}`),
+		})
+		return
+	}
+
+	m.attachExistingTargets(ctx)
+}
+
+// attachExistingTargets fetches all open targets and attaches to any that are
+// not already tracked. This catches pages that were open before Start() was called.
+func (m *Monitor) attachExistingTargets(ctx context.Context) {
+	result, err := m.send(ctx, "Target.getTargets", nil, "")
+	if err != nil {
+		return
+	}
+	var resp struct {
+		TargetInfos []cdpTargetInfo `json:"targetInfos"`
+	}
+	if err := json.Unmarshal(result, &resp); err != nil {
+		return
+	}
+	m.sessionsMu.RLock()
+	attached := make(map[string]bool, len(m.sessions))
+	for _, info := range m.sessions {
+		attached[info.targetID] = true
+	}
+	m.sessionsMu.RUnlock()
+
+	for _, ti := range resp.TargetInfos {
+		if ti.Type != targetTypePage || attached[ti.TargetID] {
+			continue
+		}
+		targetID := ti.TargetID
+		m.asyncWg.Go(func() {
+			res, err := m.send(ctx, "Target.attachToTarget", map[string]any{
+				"targetId": targetID,
+				"flatten":  true,
+			}, "")
+			if err != nil {
+				return
+			}
+			var attachResp struct {
+				SessionID string `json:"sessionId"`
+			}
+			if json.Unmarshal(res, &attachResp) == nil && attachResp.SessionID != "" {
+				m.enableDomains(ctx, attachResp.SessionID, targetTypePage)
+				_ = m.injectScript(ctx, attachResp.SessionID)
+			}
+		})
+	}
+}
+
+// restartReadLoop waits for the current readLoop to exit, then starts a new one.
+// Returns false if the context was cancelled before the restart completed.
+func (m *Monitor) restartReadLoop(ctx context.Context) bool {
+	m.lifeMu.Lock()
+	done := m.done
+	m.lifeMu.Unlock()
+
+	// Wait for old readLoop, but bail if context is cancelled (e.g. Stop called).
+	select {
+	case <-done:
+	case <-ctx.Done():
+		return false
+	}
+
+	m.lifeMu.Lock()
+	m.done = make(chan struct{})
+	m.readReady = make(chan struct{})
+	m.lifeMu.Unlock()
+
+	go m.readLoop(ctx)
+	return true
+}
+
+// subscribeToUpstream reconnects with backoff on Chrome restarts, publishing disconnect/reconnect events.
+func (m *Monitor) subscribeToUpstream(ctx context.Context) {
+	ch, cancel := m.upstreamMgr.Subscribe()
+	defer cancel()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case newURL, ok := <-ch:
+			if !ok {
+				return
+			}
+			m.handleUpstreamRestart(ctx, newURL)
+		}
+	}
+}
+
+// handleUpstreamRestart tears down the old connection, reconnects with backoff,
+// and re-initializes the CDP session. Serialized by restartMu to prevent
+// overlapping reconnects from rapid successive Chrome restarts.
+func (m *Monitor) handleUpstreamRestart(ctx context.Context, newURL string) {
+	m.restartMu.Lock()
+	defer m.restartMu.Unlock()
+
+	if ctx.Err() != nil {
+		return
+	}
+	m.publish(events.Event{
+		Ts:       time.Now().UnixMicro(),
+		Type:     EventMonitorDisconnected,
+		Category: events.CategorySystem,
+		Source:   events.Source{Kind: events.KindLocalProcess},
+		Data:     json.RawMessage(`{"reason":"` + ReasonChromeRestarted + `"}`),
+	})
+
+	startReconnect := time.Now()
+
+	m.lifeMu.Lock()
+	if m.conn != nil {
+		_ = m.conn.Close(websocket.StatusNormalClosure, "reconnecting")
+		m.conn = nil
+	}
+	m.lifeMu.Unlock()
+
+	if !m.reconnectWithBackoff(ctx, newURL) {
+		// Context cancelled means Stop() was called, not a failure.
+		if ctx.Err() == nil {
+			m.running.Store(false)
+			m.publish(events.Event{
+				Ts:       time.Now().UnixMicro(),
+				Type:     EventMonitorReconnectFailed,
+				Category: events.CategorySystem,
+				Source:   events.Source{Kind: events.KindLocalProcess},
+				Data:     json.RawMessage(`{"reason":"` + ReasonReconnectExhausted + `"}`),
+			})
+		}
+		return
+	}
+
+	// restartReadLoop waits for the old readLoop to exit before returning,
+	// so clearState runs only after the old loop has stopped touching shared state.
+	if !m.restartReadLoop(ctx) {
+		return
+	}
+	m.clearState()
+
+	m.asyncWg.Go(func() { m.initSession(ctx) })
+	reconnectDurationMs := time.Since(startReconnect).Milliseconds()
+	m.log.Info("cdpmonitor: reconnected", "url", newURL, "duration_ms", reconnectDurationMs)
+
+	m.publish(events.Event{
+		Ts:       time.Now().UnixMicro(),
+		Type:     EventMonitorReconnected,
+		Category: events.CategorySystem,
+		Source:   events.Source{Kind: events.KindLocalProcess},
+		Data:     json.RawMessage(fmt.Sprintf(`{"reconnect_duration_ms":%d}`, reconnectDurationMs)),
+	})
+}
+
+const maxReconnectAttempts = 10
+
+var reconnectBackoffs = []time.Duration{
+	250 * time.Millisecond,
+	500 * time.Millisecond,
+	1 * time.Second,
+	2 * time.Second,
+}
+
+// reconnectWithBackoff attempts to dial newURL up to maxReconnectAttempts times with exponential backoff.
+func (m *Monitor) reconnectWithBackoff(ctx context.Context, newURL string) bool {
+	for attempt := range maxReconnectAttempts {
+		if ctx.Err() != nil {
+			return false
+		}
+
+		if attempt > 0 {
+			idx := min(attempt-1, len(reconnectBackoffs)-1)
+			select {
+			case <-ctx.Done():
+				return false
+			case <-time.After(reconnectBackoffs[idx]):
+			}
+		}
+
+		conn, _, err := websocket.Dial(ctx, newURL, nil)
+		if err != nil {
+			m.log.Warn("cdpmonitor: reconnect attempt failed", "attempt", attempt+1, "max_attempts", maxReconnectAttempts, "url", newURL, "err", err)
+			continue
+		}
+		conn.SetReadLimit(wsReadLimit)
+
+		m.lifeMu.Lock()
+		m.conn = conn
+		m.lifeMu.Unlock()
+		return true
+	}
+	return false
+}
diff --git a/server/lib/cdpmonitor/monitor_test.go b/server/lib/cdpmonitor/monitor_test.go
new file mode 100644
index 00000000..f86e33c8
--- /dev/null
+++ b/server/lib/cdpmonitor/monitor_test.go
@@ -0,0 +1,357 @@
+package cdpmonitor
+
+import (
+	"context"
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// discardLogger is defined in cdp_test.go (package-level, shared across test files).
+
+func TestLifecycle(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	ec := newEventCollector()
+	upstream := newTestUpstream(srv.wsURL())
+	m := New(upstream, ec.publishFn(), 99, discardLogger)
+
+	assert.False(t, m.IsRunning(), "idle at boot")
+
+	require.NoError(t, m.Start(context.Background()))
+	assert.True(t, m.IsRunning(), "running after Start")
+	srv.readFromMonitor(t, 2*time.Second)
+
+	m.Stop()
+	assert.False(t, m.IsRunning(), "stopped after Stop")
+
+	require.NoError(t, m.Start(context.Background()))
+	assert.True(t, m.IsRunning(), "running after second Start")
+	srv.readFromMonitor(t, 2*time.Second)
+
+	require.NoError(t, m.Start(context.Background()))
+	assert.True(t, m.IsRunning(), "running after implicit restart")
+
+	m.Stop()
+	assert.False(t, m.IsRunning(), "stopped at end")
+}
+
+func TestReconnect(t *testing.T) {
+	srv1 := newTestServer(t)
+
+	upstream := newTestUpstream(srv1.wsURL())
+	ec := newEventCollector()
+	m := New(upstream, ec.publishFn(), 99, discardLogger)
+	require.NoError(t, m.Start(context.Background()))
+	defer m.Stop()
+
+	srv1.readFromMonitor(t, 2*time.Second)
+
+	srv2 := newTestServer(t)
+	defer srv2.close()
+	defer srv1.close()
+
+	upstream.notifyRestart(srv2.wsURL())
+
+	ec.waitFor(t, "monitor_disconnected", 3*time.Second)
+	srv2.readFromMonitor(t, 5*time.Second)
+
+	ev := ec.waitFor(t, "monitor_reconnected", 3*time.Second)
+	var data map[string]any
+	require.NoError(t, json.Unmarshal(ev.Data, &data))
+	_, ok := data["reconnect_duration_ms"]
+	assert.True(t, ok, "missing reconnect_duration_ms")
+}
+
+func TestScreenshot(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	m, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+
+	var captureCount atomic.Int32
+	m.screenshotFn = func(ctx context.Context, displayNum int) ([]byte, error) {
+		captureCount.Add(1)
+		return minimalPNG, nil
+	}
+
+	t.Run("capture_and_publish", func(t *testing.T) {
+		m.tryScreenshot(context.Background())
+		require.Eventually(t, func() bool { return captureCount.Load() == 1 }, 2*time.Second, 20*time.Millisecond)
+
+		ev := ec.waitFor(t, "screenshot", 2*time.Second)
+		assert.Equal(t, events.CategorySystem, ev.Category)
+		assert.Equal(t, events.KindLocalProcess, ev.Source.Kind)
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.NotEmpty(t, data["png"])
+	})
+
+	t.Run("rate_limited", func(t *testing.T) {
+		before := captureCount.Load()
+		m.tryScreenshot(context.Background())
+		time.Sleep(100 * time.Millisecond)
+		assert.Equal(t, before, captureCount.Load(), "should be rate-limited within 2s")
+	})
+
+	t.Run("captures_after_cooldown", func(t *testing.T) {
+		m.lastScreenshotAt.Store(time.Now().Add(-3 * time.Second).UnixMilli())
+		before := captureCount.Load()
+		m.tryScreenshot(context.Background())
+		require.Eventually(t, func() bool { return captureCount.Load() > before }, 2*time.Second, 20*time.Millisecond)
+	})
+}
+
+// TestFailPendingCommandsUnblocksSend verifies that clearState (called during
+// reconnect) unblocks any goroutine blocked in send() by delivering an error.
+func TestFailPendingCommandsUnblocksSend(t *testing.T) {
+	ec := newEventCollector()
+	upstream := newTestUpstream("ws://127.0.0.1:0")
+	m := New(upstream, ec.publishFn(), 0, discardLogger)
+
+	// Pre-register a fake pending command channel as if send() had registered it.
+	id := int64(42)
+	ch := make(chan cdpMessage, 1)
+	m.pendMu.Lock()
+	m.pending[id] = ch
+	m.pendMu.Unlock()
+
+	// failPendingCommands should deliver an error message to ch without blocking.
+	done := make(chan struct{})
+	go func() {
+		m.failPendingCommands()
+		close(done)
+	}()
+
+	select {
+	case msg := <-ch:
+		require.NotNil(t, msg.Error, "expected error response from failPendingCommands")
+		assert.Equal(t, -1, msg.Error.Code)
+	case <-time.After(2 * time.Second):
+		t.Fatal("failPendingCommands did not unblock the pending channel")
+	}
+	<-done
+}
+
+// TestInitSessionAutoAttachFailure verifies that a monitor_init_failed event is
+// published (and the monitor logs the failure) when Target.setAutoAttach returns
+// an error.
+func TestInitSessionAutoAttachFailure(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	ec := newEventCollector()
+	upstream := newTestUpstream(srv.wsURL())
+	m := New(upstream, ec.publishFn(), 99, discardLogger)
+	require.NoError(t, m.Start(context.Background()))
+	defer m.Stop()
+
+	stopResponder := make(chan struct{})
+	defer close(stopResponder)
+
+	go listenAndRespond(srv, stopResponder, func(msg cdpMessage) any {
+		if msg.Method == "Target.setAutoAttach" {
+			return map[string]any{
+				"id":    msg.ID,
+				"error": map[string]any{"code": -32601, "message": "Method not found"},
+			}
+		}
+		return nil
+	})
+
+	ec.waitFor(t, EventMonitorInitFailed, 3*time.Second)
+}
+
+func TestAutoAttach(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	ec := newEventCollector()
+	upstream := newTestUpstream(srv.wsURL())
+	m := New(upstream, ec.publishFn(), 99, discardLogger)
+	require.NoError(t, m.Start(context.Background()))
+	defer m.Stop()
+
+	msg := srv.readFromMonitor(t, 3*time.Second)
+	assert.Equal(t, "Target.setAutoAttach", msg.Method)
+
+	var params struct {
+		AutoAttach             bool `json:"autoAttach"`
+		WaitForDebuggerOnStart bool `json:"waitForDebuggerOnStart"`
+		Flatten                bool `json:"flatten"`
+	}
+	require.NoError(t, json.Unmarshal(msg.Params, &params))
+	assert.True(t, params.AutoAttach)
+	assert.False(t, params.WaitForDebuggerOnStart)
+	assert.True(t, params.Flatten)
+
+	stopResponder := make(chan struct{})
+	go listenAndRespond(srv, stopResponder, nil)
+	defer close(stopResponder)
+	srv.sendToMonitor(t, map[string]any{"id": msg.ID, "result": map[string]any{}})
+
+	srv.sendToMonitor(t, map[string]any{
+		"method": "Target.attachedToTarget",
+		"params": map[string]any{
+			"sessionId":  "session-abc",
+			"targetInfo": map[string]any{"targetId": "target-xyz", "type": "page", "url": "https://example.com"},
+		},
+	})
+	require.Eventually(t, func() bool {
+		m.sessionsMu.RLock()
+		defer m.sessionsMu.RUnlock()
+		_, ok := m.sessions["session-abc"]
+		return ok
+	}, 2*time.Second, 50*time.Millisecond, "session not stored")
+
+	m.sessionsMu.RLock()
+	info := m.sessions["session-abc"]
+	m.sessionsMu.RUnlock()
+	assert.Equal(t, "target-xyz", info.targetID)
+	assert.Equal(t, "page", info.targetType)
+}
+
+func TestAttachExistingTargets(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+
+	responder := func(msg cdpMessage) any {
+		switch msg.Method {
+		case "Target.getTargets":
+			return map[string]any{
+				"id": msg.ID,
+				"result": map[string]any{
+					"targetInfos": []any{
+						map[string]any{"targetId": "existing-1", "type": "page", "url": "https://preexisting.example.com"},
+					},
+				},
+			}
+		case "Target.attachToTarget":
+			srv.sendToMonitor(t, map[string]any{
+				"method": "Target.attachedToTarget",
+				"params": map[string]any{
+					"sessionId":  "session-existing-1",
+					"targetInfo": map[string]any{"targetId": "existing-1", "type": "page", "url": "https://preexisting.example.com"},
+				},
+			})
+			return map[string]any{"id": msg.ID, "result": map[string]any{"sessionId": "session-existing-1"}}
+		}
+		return nil
+	}
+
+	m, _, cleanup := startMonitor(t, srv, responder)
+	defer cleanup()
+
+	require.Eventually(t, func() bool {
+		m.sessionsMu.RLock()
+		defer m.sessionsMu.RUnlock()
+		_, ok := m.sessions["session-existing-1"]
+		return ok
+	}, 3*time.Second, 50*time.Millisecond, "existing target not auto-attached")
+
+	m.sessionsMu.RLock()
+	info := m.sessions["session-existing-1"]
+	m.sessionsMu.RUnlock()
+	assert.Equal(t, "existing-1", info.targetID)
+}
+
+// TestRedirectCounter verifies that redirect hops (same requestId, multiple
+// requestWillBeSent) do not double-increment netPending, which would permanently
+// block network_idle.
+func TestRedirectCounter(t *testing.T) {
+	m, ec := newComputedMonitor(t)
+	navigateMonitor(m, "https://example.com")
+
+	// First requestWillBeSent — genuine new request.
+	p1, _ := json.Marshal(map[string]any{
+		"requestId":    "r-redirect",
+		"resourceType": "Document",
+		"request":      map[string]any{"method": "GET", "url": "https://example.com/old"},
+		"initiator":    map[string]any{"type": "other"},
+	})
+	m.handleNetworkRequest(p1, "s1")
+
+	// Second requestWillBeSent with the same requestId — this is the redirect hop.
+	p2, _ := json.Marshal(map[string]any{
+		"requestId":    "r-redirect",
+		"resourceType": "Document",
+		"request":      map[string]any{"method": "GET", "url": "https://example.com/new"},
+		"initiator":    map[string]any{"type": "other"},
+	})
+	m.handleNetworkRequest(p2, "s1")
+
+	// Only one loadingFinished fires per redirect chain.
+	p3, _ := json.Marshal(map[string]any{"requestId": "r-redirect"})
+	m.handleLoadingFinished(p3, "s1")
+
+	// If netPending was double-incremented, network_idle would never fire.
+	ec.waitFor(t, "network_idle", 2*time.Second)
+}
+
+// TestSubframeNavigationNoReset verifies that a frameNavigated event with a
+// non-empty parentId does not reset computed state (netPending, timers, etc.).
+func TestSubframeNavigationNoReset(t *testing.T) {
+	m, ec := newComputedMonitor(t)
+	navigateMonitor(m, "https://example.com") // top-level nav, sets mainSessionID
+
+	// Start a request on the main frame.
+	simulateRequest(m, "main-req")
+
+	// An iframe navigates — should not reset state or clear pendingRequests.
+	iframeNav, _ := json.Marshal(map[string]any{
+		"frame": map[string]any{
+			"id":       "iframe-frame",
+			"parentId": "top-frame",
+			"url":      "https://iframe.example.com",
+		},
+	})
+	m.handleFrameNavigated(iframeNav, "s1")
+
+	// mainSessionID should still be "s1", not reset by the subframe nav.
+	assert.Equal(t, "s1", m.mainSessionID.Load(), "mainSessionID should not change on subframe nav")
+
+	// Finishing the main request should still drive network_idle (state not reset).
+	simulateFinished(m, "main-req")
+	ec.waitFor(t, "network_idle", 2*time.Second)
+}
+
+func TestSubframeLifecycleIgnored(t *testing.T) {
+	t.Run("subframe_dom_content_loaded_does_not_advance_state", func(t *testing.T) {
+		m, ec := newComputedMonitor(t)
+		navigateMonitor(m, "https://example.com") // sets mainSessionID = "s1"
+
+		// Fire domContentLoaded from an iframe session, not the main frame.
+		m.handleDOMContentLoaded(json.RawMessage(`{}`), "iframe-session")
+
+		// Now fire the real main-frame domContentLoaded + the rest of the conditions.
+		simulateRequest(m, "r1")
+		simulateFinished(m, "r1")
+		m.handleLoadEventFired(json.RawMessage(`{}`), "s1")
+		// navigation_settled requires navDOMLoaded; if the iframe event had set it,
+		// the event might fire without the main-frame DOMContentLoaded arriving.
+		// Assert it does NOT fire yet (iframe set navDOMLoaded but main frame hasn't).
+		ec.assertNone(t, "navigation_settled", 1500*time.Millisecond)
+	})
+
+	t.Run("subframe_load_event_does_not_start_layout_timer", func(t *testing.T) {
+		m, ec := newComputedMonitor(t)
+		navigateMonitor(m, "https://example.com")
+
+		// Subframe fires loadEventFired — should not start the layout_settled timer.
+		m.handleLoadEventFired(json.RawMessage(`{}`), "iframe-session")
+		ec.assertNone(t, "layout_settled", 1500*time.Millisecond)
+
+		// Main frame fires — timer should start now.
+		t0 := time.Now()
+		m.handleLoadEventFired(json.RawMessage(`{}`), "s1")
+		ec.waitFor(t, "layout_settled", 3*time.Second)
+		assert.GreaterOrEqual(t, time.Since(t0).Milliseconds(), int64(900), "fired too early")
+	})
+}
diff --git a/server/lib/cdpmonitor/screenshot.go b/server/lib/cdpmonitor/screenshot.go
new file mode 100644
index 00000000..2a06d406
--- /dev/null
+++ b/server/lib/cdpmonitor/screenshot.go
@@ -0,0 +1,99 @@
+package cdpmonitor
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"os/exec"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/events"
+)
+
+// tryScreenshot fires a screenshot if the 2s rate-limit window has elapsed.
+// lastScreenshotAt CAS enforces the window; screenshotInFlight CAS prevents
+// overlapping captures when captureScreenshot outlasts the 2s window (timeout is 10s).
+func (m *Monitor) tryScreenshot(ctx context.Context) {
+	now := time.Now().UnixMilli()
+	last := m.lastScreenshotAt.Load()
+	if now-last < 2000 {
+		return
+	}
+	if !m.lastScreenshotAt.CompareAndSwap(last, now) {
+		return
+	}
+	if !m.screenshotInFlight.CompareAndSwap(false, true) {
+		return
+	}
+	m.asyncWg.Go(func() {
+		defer m.screenshotInFlight.Store(false)
+		m.captureScreenshot(ctx)
+	})
+}
+
+const screenshotTimeout = 10 * time.Second
+
+// captureScreenshot takes a screenshot via ffmpeg x11grab (or the screenshotFn
+// seam in tests), optionally downscales it, and publishes a screenshot event.
+func (m *Monitor) captureScreenshot(parentCtx context.Context) {
+	ctx, cancel := context.WithTimeout(parentCtx, screenshotTimeout)
+	defer cancel()
+	var pngBytes []byte
+	var err error
+
+	if m.screenshotFn != nil {
+		pngBytes, err = m.screenshotFn(ctx, m.displayNum)
+	} else {
+		pngBytes, err = captureViaFFmpeg(ctx, m.displayNum, 1)
+	}
+	if err != nil {
+		m.log.Warn("cdpmonitor: screenshot capture failed", "err", err)
+		return
+	}
+
+	// Downscale if base64 output would exceed ~972KB (~729KB raw × 4/3 base64 inflation).
+	const rawThreshold = 729 * 1024
+	for scale := 2; len(pngBytes) > rawThreshold && scale <= 16 && m.screenshotFn == nil; scale *= 2 {
+		pngBytes, err = captureViaFFmpeg(ctx, m.displayNum, scale)
+		if err != nil {
+			m.log.Warn("cdpmonitor: screenshot downscale failed", "scale", scale, "err", err)
+			return
+		}
+	}
+
+	encoded := base64.StdEncoding.EncodeToString(pngBytes)
+	data, _ := json.Marshal(map[string]string{screenshotDataKey: encoded})
+
+	m.publish(events.Event{
+		Ts:       time.Now().UnixMicro(),
+		Type:     EventScreenshot,
+		Category: events.CategorySystem,
+		Source:   events.Source{Kind: events.KindLocalProcess},
+		Data:     data,
+	})
+}
+
+// captureViaFFmpeg runs ffmpeg x11grab to capture a PNG screenshot.
+// If divisor > 1, a scale filter is applied to reduce the output size.
+func captureViaFFmpeg(ctx context.Context, displayNum, divisor int) ([]byte, error) {
+	args := []string{
+		"-f", "x11grab",
+		"-i", fmt.Sprintf(":%d", displayNum),
+		"-vframes", "1",
+	}
+	if divisor > 1 {
+		args = append(args, "-vf", fmt.Sprintf("scale=iw/%d:ih/%d", divisor, divisor))
+	}
+	args = append(args, "-f", "image2", "pipe:1")
+
+	var out, stderr bytes.Buffer
+	cmd := exec.CommandContext(ctx, "ffmpeg", args...)
+	cmd.Stdout = &out
+	cmd.Stderr = &stderr
+	if err := cmd.Run(); err != nil {
+		return nil, fmt.Errorf("%w: %s", err, stderr.String())
+	}
+	return out.Bytes(), nil
+}