From ab9218e52faf551b0ae4176688e0936adc230efd Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:01:01 -0700 Subject: [PATCH 1/7] fix(source): stop re-sending unchanged pushes and the 60s watch budget A watch-mode source re-sent the full envelope to every sink on each 30s baseline tick, even when nothing had changed, and each watch-mode push was capped at 60s, below the 5-minute SyncClient timeout, so slow links failed with context deadline exceeded. The source now remembers, per sink, the hash of the last envelope it delivered (Sequence and cookie access times excluded) and skips a sink that already holds the same payload, re-sending at least every 10 minutes as a heartbeat. Only a successful POST counts as delivered, so a failed sink is always retried. Watch-mode pushes get the same budget as --once (SyncClient timeout plus 30s), and the watcher skips a trigger while a push is still running so slow pushes cannot pile up. Reported by @Dannyxlm in #149. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- docs/faq.md | 2 + internal/cli/source.go | 152 +++++++++++++++++-- internal/cli/source_test.go | 253 +++++++++++++++++++++++++++++++ internal/cli/teammates_test.go | 2 + internal/watcher/watcher.go | 52 +++++-- internal/watcher/watcher_test.go | 126 +++++++++++++++ 6 files changed, 561 insertions(+), 26 deletions(-) diff --git a/docs/faq.md b/docs/faq.md index 9847865..1761ce1 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -32,6 +32,8 @@ domains: [] The 1.0 release does NOT change this default in code. A later release may flip the default, which would be a breaking change. +How often does the source send? `source --watch` checks every 30 seconds and on every cookie change, but it sends to a sink only when the payload changed since that sink last accepted it, or 10 minutes have passed. A sink whose last push failed gets the next one regardless. Run with `--verbose` to see the skipped sends. + ## Will syncing cookies log me out of sites on the source machine? No. The source reads cookies with `immutable=1` (Chrome's recommended read-only flag), and `agentcookie source` never writes to the source's Cookies SQLite. The only writes happen on the sink. diff --git a/internal/cli/source.go b/internal/cli/source.go index 44fed54..cbf2a57 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -3,6 +3,7 @@ package cli import ( "bytes" "context" + "crypto/sha256" "encoding/json" "errors" "fmt" @@ -11,6 +12,7 @@ import ( "net/url" "os" "path/filepath" + "slices" "strings" "sync" "time" @@ -96,6 +98,87 @@ const ( sinkBackoffMax = 30 * time.Minute ) +// sinkHeartbeat is how long an unchanged payload goes without being re-sent +// to a sink that already received it. doctor only warns after 24h without a +// push, so 10 minutes keeps status fresh while an idle source stops sending +// the same envelope every 30s (#149). +const sinkHeartbeat = 10 * time.Minute + +// sinkDelivery is the last payload a sink accepted: the envelope hash (see +// envelopeHash) and when it was sent. +type sinkDelivery struct { + hash [sha256.Size]byte + at time.Time +} + +// sinkDeliveryLog remembers, per sink (peer + URL), the last successful +// delivery of this process. It lives in memory only: a restarted source +// sends everything once, which is what --once and a fresh daemon want. +type sinkDeliveryLog struct { + mu sync.Mutex + m map[string]sinkDelivery +} + +func newSinkDeliveryLog() *sinkDeliveryLog { + return &sinkDeliveryLog{m: map[string]sinkDelivery{}} +} + +// sinkDeliveries is the source process's delivery log. Tests replace it so +// one test's deliveries do not suppress another's. +var sinkDeliveries = newSinkDeliveryLog() + +func sinkDeliveryKey(peer, rawURL string) string { + return peer + "\x00" + rawURL +} + +// unchanged reports whether the sink already received hash less than +// sinkHeartbeat before now, and if so when. +func (l *sinkDeliveryLog) unchanged(peer, rawURL string, hash [sha256.Size]byte, now time.Time) (time.Time, bool) { + l.mu.Lock() + defer l.mu.Unlock() + d, ok := l.m[sinkDeliveryKey(peer, rawURL)] + if !ok || d.hash != hash || now.Sub(d.at) >= sinkHeartbeat { + return time.Time{}, false + } + return d.at, true +} + +// record notes a successful delivery. Only a sink that accepted the POST is +// recorded, so a failed push is always re-sent. +func (l *sinkDeliveryLog) record(peer, rawURL string, hash [sha256.Size]byte, now time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + l.m[sinkDeliveryKey(peer, rawURL)] = sinkDelivery{hash: hash, at: now} +} + +// envelopeHash identifies an envelope's content for the unchanged check. +// Sequence is zeroed: every real POST needs a fresh one for the sink's +// replay tracker, so it must not make identical content look new. +// +// Cookie LastAccessUTC is zeroed too, in the hash only (the sent payload +// keeps it). Chrome bumps last_access_utc whenever a request sends a cookie, +// throttled to once a minute per cookie (CookieMonster's 60s access-update +// threshold), and its SQLite store commits pending writes every 30s. With +// any tab open, some access time changes between most 30s baseline ticks, +// which would defeat the check. The sink only uses access times to order +// live CDP injection, and the heartbeat carries fresh ones every 10 minutes. +func envelopeHash(env protocol.SyncEnvelope) ([sha256.Size]byte, error) { + env.Sequence = 0 + cookies := make([]chrome.Cookie, len(env.Cookies)) + for i, c := range env.Cookies { + c.LastAccessUTC = 0 + cookies[i] = c + } + env.Cookies = cookies + h := sha256.New() + if err := json.NewEncoder(h).Encode(env); err != nil { + return [sha256.Size]byte{}, err + } + var sum [sha256.Size]byte + copy(sum[:], h.Sum(nil)) + return sum, nil +} + // errSinkAsleep marks a sink whose Tailscale peer is offline. The source // skips it without dialing and records it as asleep, not as a failure. var errSinkAsleep = errors.New("sink peer is offline") @@ -220,17 +303,7 @@ func runSource(cmd *cobra.Command, args []string) error { } if sourceOnce { - // --once mode: bound the whole push by SyncClient's timeout - // plus a small slack for envelope packing. Pre-v0.12 this was - // hardcoded at 60s, which was tight even for v0.10-shape - // payloads. The inner HTTP request also bounds itself; this - // outer cancel is the belt to the request's suspenders. - // - // Fan-out POSTs run concurrently, each under its own per-sink - // ClientTimeout, so one per-sink timeout plus slack covers every - // sink: a dead sink cannot starve the healthy ones. - perSink := httpserver.Defaults(httpserver.SyncClient).ClientTimeout - ctx, cancel := context.WithTimeout(cmd.Context(), perSink+30*time.Second) + ctx, cancel := context.WithTimeout(cmd.Context(), sourcePushBudget()) defer cancel() _, err := push(ctx) return err @@ -257,6 +330,7 @@ func runSource(cmd *cobra.Command, args []string) error { LocalStorageDir: sourceBrowser.LocalStorageLevelDB(cfg.Browser.Profile), IndexedDBDir: sourceBrowser.IndexedDBDir(cfg.Browser.Profile), Push: push, + PushTimeout: sourcePushBudget(), OnEvent: func(ev watcher.Event) { if sourceVerbose { fmt.Fprintf(os.Stderr, "agentcookie source --watch: %s\n", ev.String()) @@ -310,6 +384,20 @@ func runSource(cmd *cobra.Command, args []string) error { return w.Run(cmd.Context()) } +// sourcePushBudget bounds one whole source push cycle, in --once and in +// --watch alike: SyncClient's per-sink timeout plus a small slack for +// envelope packing. Pre-v0.12 --once was hardcoded at 60s, which was tight +// even for v0.10-shape payloads, and --watch kept that 60s until #149. The +// inner HTTP request also bounds itself; this outer cancel is the belt to the +// request's suspenders. +// +// Fan-out POSTs run concurrently, each under its own per-sink ClientTimeout, +// so one per-sink timeout plus slack covers every sink: a dead sink cannot +// starve the healthy ones. +func sourcePushBudget() time.Duration { + return httpserver.Defaults(httpserver.SyncClient).ClientTimeout + 30*time.Second +} + // newSourceState builds the in-memory SourceState for a source run. Per-sink // records from the previous run's state file (backoff window, consecutive // failures, asleep marker, and history) carry over, so a restart does not @@ -514,6 +602,8 @@ func recordSourcePushResult( switch { case !r.SkippedUntil.IsZero(): // Skipped inside its backoff window: no attempt, nothing to record. + case r.Unchanged: + // Already holds this payload: no attempt, nothing to record. case r.Asleep: ps.Asleep = true ps.LastAsleepAt = now @@ -558,8 +648,9 @@ func recordSourcePushResult( } // sinkResult is one sink's outcome from a single fan-out push. A sink that -// was not dialed has Asleep (Tailscale reports its peer offline) or -// SkippedUntil (inside its backoff window) set and a nil Err. +// was not dialed has Asleep (Tailscale reports its peer offline), +// SkippedUntil (inside its backoff window), or Unchanged (it already holds +// this payload, inside the heartbeat) set and a nil Err. type sinkResult struct { Peer string URL string @@ -567,6 +658,9 @@ type sinkResult struct { Err error Asleep bool SkippedUntil time.Time + Unchanged bool + // DeliveredAt is the last delivery of this payload, for an Unchanged sink. + DeliveredAt time.Time } // sinkURLList renders sink URLs for a log line. @@ -730,6 +824,9 @@ func pushOnce( if os.Getenv("AGENTCOOKIE_SYNC_INDEXEDDB") == "1" { if it, sk, err := chromedirsync.Pack(sourceBrowser.IndexedDBDir(cfg.Browser.Profile), 5*1024*1024); err == nil { idbTarball = it + // Pack builds this list from a map; sort it so an unchanged + // profile hashes the same every cycle. + slices.Sort(sk) idbSkipped = sk } else if !errors.Is(err, chromedirsync.ErrSourceMissing) { fmt.Fprintf(os.Stderr, "agentcookie source: indexedDB pack failed (%v); continuing without it\n", err) @@ -755,13 +852,19 @@ func pushOnce( return nil, dbsc, fmt.Errorf("marshal envelope: %w", err) } pullPayloadCache.StoreIfCurrent(pullGen, payload) + hash, err := envelopeHash(envelope) + if err != nil { + return nil, dbsc, fmt.Errorf("hash envelope: %w", err) + } // Fan out: read and filtering above happened once; only sealing and // transport repeat per sink. Each sink is sealed with its own key and // POSTed concurrently under its own timeout, so a dead sink cannot hold // up the others. A per-sink failure (missing key, seal error, // URL-resolution, or POST) is isolated to that sink and recorded. A sink - // inside its backoff window is skipped without an attempt, and a sink - // whose peer Tailscale reports offline is skipped as asleep. + // inside its backoff window is skipped without an attempt, a sink that + // already received this exact payload less than sinkHeartbeat ago is + // skipped as unchanged, and a sink whose peer Tailscale reports offline + // is skipped as asleep. // // A legacy push peer migrated to teammate mode is never dialed again (it // pulls instead): it is dropped silently, not recorded as a failure, and @@ -798,6 +901,11 @@ func pushOnce( continue } } + if at, ok := sinkDeliveries.unchanged(sink.Peer, sink.URL, hash, now); ok { + results[i].Unchanged = true + results[i].DeliveredAt = at + continue + } if tailnetIPOffline(sink.URL) { results[i].Asleep = true continue @@ -838,6 +946,13 @@ func pushOnce( case !r.SkippedUntil.IsZero(): rep["backoff_until"] = r.SkippedUntil.UTC().Format(time.RFC3339) humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] skipped: backing off after failures until %s", r.URL, r.SkippedUntil.Format(time.RFC3339))) + case r.Unchanged: + rep["unchanged"] = true + // Every baseline tick of an idle source lands here, so only + // --verbose logs it. + if verbose { + humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] skipped: unchanged since %s, heartbeat due by %s", r.URL, r.DeliveredAt.Format(time.RFC3339), r.DeliveredAt.Add(sinkHeartbeat).Format(time.RFC3339))) + } case r.Asleep: rep["asleep"] = true humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] skipped: Tailscale reports the peer offline (asleep)", r.URL)) @@ -846,6 +961,7 @@ func pushOnce( humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] push failed: %v", r.URL, r.Err)) default: delivered++ + sinkDeliveries.record(r.Peer, r.URL, hash, now) rep["posted"] = true humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] posted %d cookies, sink replied: %s%s", r.URL, r.Count, replies[i], dbscNote(dbsc))) } @@ -854,7 +970,11 @@ func pushOnce( result["sinks"] = sinkReports result["posted"] = delivered > 0 && len(aggregateFailedLabels(results)) == 0 - _ = emit(result, strings.Join(humanLines, "\n")+"\n") + human := "" + if len(humanLines) > 0 { + human = strings.Join(humanLines, "\n") + "\n" + } + _ = emit(result, human) return results, dbsc, nil } diff --git a/internal/cli/source_test.go b/internal/cli/source_test.go index 60fc892..7dc93c4 100644 --- a/internal/cli/source_test.go +++ b/internal/cli/source_test.go @@ -22,6 +22,7 @@ import ( "github.com/spf13/cobra" "github.com/mvanhorn/agentcookie/internal/chrome" + "github.com/mvanhorn/agentcookie/internal/cli/httpserver" "github.com/mvanhorn/agentcookie/internal/config" "github.com/mvanhorn/agentcookie/internal/protocol" "github.com/mvanhorn/agentcookie/internal/secretsbus" @@ -328,9 +329,13 @@ domains: - pattern: "%.blocked.com" `) + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + if _, err := fx.push(); err != nil { t.Fatalf("first push: %v", err) } + // Past the heartbeat, so the unchanged payload is POSTed again. + *now = now.Add(sinkHeartbeat) if _, err := fx.push(); err != nil { t.Fatalf("second push: %v", err) } @@ -367,6 +372,11 @@ func newSourcePushFixture(t *testing.T, cookies []chrome.Cookie) *sourcePushFixt return rawURL, nil })) + // Each test starts with no remembered deliveries. + prevDeliveries := sinkDeliveries + sinkDeliveries = newSinkDeliveryLog() + t.Cleanup(func() { sinkDeliveries = prevDeliveries }) + key := []byte("0123456789abcdef") dbPath := filepath.Join(t.TempDir(), "Cookies") seedSourceCookiesDB(t, dbPath, cookies, key) @@ -401,6 +411,13 @@ func (f *sourcePushFixture) push() (int, error) { return pushWithFreshBlocklist(context.Background(), f.cfg, f.key, false, false, false, f.srcState, nil) } +// forgetDeliveries clears the unchanged-payload log, so the next push POSTs +// to every eligible sink even if the payload has not changed. For tests +// that count POSTs across repeated identical pushes for another reason. +func (f *sourcePushFixture) forgetDeliveries() { + sinkDeliveries = newSinkDeliveryLog() +} + func (f *sourcePushFixture) batchCount() int { return f.capture.batchCount() } @@ -827,7 +844,10 @@ func TestSourcePushBackoffScheduleCapsAndResetsOnSuccess(t *testing.T) { 30 * time.Second, time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 16 * time.Minute, 30 * time.Minute, 30 * time.Minute, } + // The payload never changes; forget deliveries so the healthy sink is + // POSTed every cycle and only backoff decides who is skipped. for i, delay := range want { + fx.forgetDeliveries() if _, err := fx.push(); err == nil { t.Fatalf("failure %d: expected push error", i+1) } @@ -844,6 +864,7 @@ func TestSourcePushBackoffScheduleCapsAndResetsOnSuccess(t *testing.T) { // A push inside the window skips the bad sink but still serves the // healthy one, and is not an error. *now = now.Add(delay - time.Second) + fx.forgetDeliveries() if _, err := fx.push(); err != nil { t.Fatalf("failure %d: push inside backoff window should skip, got %v", i+1, err) } @@ -869,6 +890,7 @@ func TestSourcePushBackoffScheduleCapsAndResetsOnSuccess(t *testing.T) { t.Fatalf("success should reset backoff, got %+v", ps) } fx.capture.failURL = urlBad + fx.forgetDeliveries() _, _ = fx.push() if got := sinkStateFor(fx.srcState, urlBad).BackoffUntil.Sub(*now); got != want[0] { t.Fatalf("first failure after reset: backoff = %s, want %s", got, want[0]) @@ -1154,3 +1176,234 @@ func TestSourcePushTailnetIPSinkDialsWhenStatusUnavailable(t *testing.T) { t.Fatalf("expected a normal push record, got %+v", ps) } } + +func TestSourcePushBudgetIsSyncClientTimeoutPlusSlack(t *testing.T) { + want := httpserver.Defaults(httpserver.SyncClient).ClientTimeout + 30*time.Second + if got := sourcePushBudget(); got != want { + t.Fatalf("sourcePushBudget() = %s, want %s", got, want) + } + if got := sourcePushBudget(); got <= 60*time.Second { + t.Fatalf("sourcePushBudget() = %s, want more than the watcher's 60s default", got) + } +} + +// --- Unchanged-payload dedup with heartbeat (#149) --- + +func TestSourcePushSkipsUnchangedPayloadUntilHeartbeat(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + sinkURL := fx.cfg.Sink.URL + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err != nil { + t.Fatalf("first push: %v", err) + } + // Baseline ticks every 30s for 9 minutes with nothing changed. + for range 18 { + *now = now.Add(30 * time.Second) + n, err := fx.push() + if err != nil { + t.Fatalf("unchanged push: %v", err) + } + if n != 0 { + t.Fatalf("unchanged push reported %d cookies posted, want 0", n) + } + } + if got := fx.capture.attemptsTo(sinkURL); got != 1 { + t.Fatalf("POSTs inside the heartbeat window = %d, want 1", got) + } + ps := sinkStateFor(fx.srcState, sinkURL) + if ps == nil || ps.TotalPushes != 1 { + t.Fatalf("unchanged skips must not record pushes, got %+v", ps) + } + if !ps.LastPush.Equal(time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) { + t.Errorf("LastPush = %s, want the first delivery", ps.LastPush) + } + + // 10 minutes after the last delivery the heartbeat re-sends it. + *now = time.Date(2026, 10, 11, 12, 10, 0, 0, time.UTC) + if _, err := fx.push(); err != nil { + t.Fatalf("heartbeat push: %v", err) + } + if got := fx.capture.attemptsTo(sinkURL); got != 2 { + t.Fatalf("POSTs after the heartbeat = %d, want 2", got) + } + if got := sinkStateFor(fx.srcState, sinkURL).TotalPushes; got != 2 { + t.Errorf("TotalPushes after heartbeat = %d, want 2", got) + } +} + +func TestSourcePushChangedCookiePostsImmediately(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v1", Path: "/"}, + }) + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err != nil { + t.Fatalf("first push: %v", err) + } + seedSourceCookiesDB(t, fx.cfg.Chrome.DBPath, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v2", Path: "/"}, + }, fx.key) + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("second push: %v", err) + } + if got := fx.batchCount(); got != 2 { + t.Fatalf("POSTs = %d, want 2 after a cookie value changed", got) + } + if got := fx.capture.batchAt(1)[0].Value; got != "v2" { + t.Errorf("second POST cookie value = %q, want v2", got) + } +} + +func TestSourcePushLastAccessChangeAloneIsUnchanged(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/", LastAccessUTC: 13370000000000000}, + }) + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err != nil { + t.Fatalf("first push: %v", err) + } + seedSourceCookiesDB(t, fx.cfg.Chrome.DBPath, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/", LastAccessUTC: 13370000060000000}, + }, fx.key) + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("second push: %v", err) + } + if got := fx.batchCount(); got != 1 { + t.Fatalf("POSTs = %d, want 1: a last-access bump alone is not a change", got) + } + + // The heartbeat still carries the current access time. + *now = now.Add(10 * time.Minute) + if _, err := fx.push(); err != nil { + t.Fatalf("heartbeat push: %v", err) + } + if got := fx.batchCount(); got != 2 { + t.Fatalf("POSTs after heartbeat = %d, want 2", got) + } + if got := fx.capture.batchAt(1)[0].LastAccessUTC; got != 13370000060000000 { + t.Errorf("heartbeat LastAccessUTC = %d, want the updated value", got) + } +} + +func TestSourcePushFailedDeliveryIsResent(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + sinkURL := fx.cfg.Sink.URL + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + fx.capture.failURL = sinkURL + + if _, err := fx.push(); err == nil { + t.Fatal("first push should fail") + } + fx.capture.failURL = "" + // Past the 30s backoff window, same payload. + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("retry push: %v", err) + } + if got := fx.capture.attemptsTo(sinkURL); got != 2 { + t.Fatalf("POST attempts = %d, want 2: a failed POST never counts as delivered", got) + } + if got := fx.batchCount(); got != 1 { + t.Fatalf("delivered batches = %d, want 1", got) + } +} + +func TestSourcePushResendsOnlyToFailedSink(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + urlBad := "http://sink-bad.test/sync" + urlGood := "http://sink-good.test/sync" + fx.cfg.Sink = config.SinkRef{} + fx.cfg.Sinks = []config.SinkTarget{{URL: urlBad}, {URL: urlGood}} + fx.capture.failURL = urlBad + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err == nil { + t.Fatal("first push should report the failed sink") + } + fx.capture.failURL = "" + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("second push: %v", err) + } + if got := fx.capture.attemptsTo(urlBad); got != 2 { + t.Errorf("failed sink attempts = %d, want 2 (resent)", got) + } + if got := fx.capture.attemptsTo(urlGood); got != 1 { + t.Errorf("delivered sink attempts = %d, want 1 (unchanged, skipped)", got) + } +} + +func TestSourcePushRealPostsCarryIncreasingSequence(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v0", Path: "/"}, + }) + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + // Mix changed payloads, unchanged skips and a heartbeat. + steps := []struct { + value string + advance time.Duration + }{ + {"v0", 0}, {"v0", 30 * time.Second}, {"v1", 30 * time.Second}, + {"v1", 30 * time.Second}, {"v1", 10 * time.Minute}, {"v2", 30 * time.Second}, + } + for i, s := range steps { + *now = now.Add(s.advance) + seedSourceCookiesDB(t, fx.cfg.Chrome.DBPath, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: s.value, Path: "/"}, + }, fx.key) + if _, err := fx.push(); err != nil { + t.Fatalf("push %d: %v", i, err) + } + } + if got := fx.batchCount(); got != 4 { + t.Fatalf("real POSTs = %d, want 4", got) + } + for i := 1; i < fx.batchCount(); i++ { + prev, cur := fx.capture.envelopeAt(i-1).Sequence, fx.capture.envelopeAt(i).Sequence + if cur <= prev { + t.Fatalf("POST %d Sequence %d not greater than previous %d", i, cur, prev) + } + } +} + +func TestSourcePushUnchangedSkipLeavesFailureStateAlone(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + sinkURL := fx.cfg.Sink.URL + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err != nil { + t.Fatalf("first push: %v", err) + } + before := *sinkStateFor(fx.srcState, sinkURL) + srcBefore := fx.srcState.TotalFailures + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("unchanged push should not be an error, got %v", err) + } + after := *sinkStateFor(fx.srcState, sinkURL) + if after.TotalFailures != before.TotalFailures || after.ConsecutiveFailures != before.ConsecutiveFailures { + t.Errorf("unchanged skip touched failure counts: before %+v after %+v", before, after) + } + if !after.BackoffUntil.Equal(before.BackoffUntil) { + t.Errorf("unchanged skip touched backoff: %s -> %s", before.BackoffUntil, after.BackoffUntil) + } + if after.TotalPushes != before.TotalPushes || !after.LastPush.Equal(before.LastPush) { + t.Errorf("unchanged skip recorded a delivery: before %+v after %+v", before, after) + } + if fx.srcState.TotalFailures != srcBefore { + t.Errorf("source TotalFailures = %d, want %d", fx.srcState.TotalFailures, srcBefore) + } +} diff --git a/internal/cli/teammates_test.go b/internal/cli/teammates_test.go index 7ae43bf..74b34a2 100644 --- a/internal/cli/teammates_test.go +++ b/internal/cli/teammates_test.go @@ -207,6 +207,7 @@ func TestLegacySinkMigratedWhenListed(t *testing.T) { } before := len(fx.capture.postedURLs()) for range 3 { + fx.forgetDeliveries() if _, err := fx.push(); err != nil { t.Fatalf("push: %v", err) } @@ -321,6 +322,7 @@ func TestPushHeldWhileTeammateStateUnavailable(t *testing.T) { t.Fatalf("list %q: SuppressPush(instinct) = false with the state store unavailable", list) } before := len(fx.capture.postedURLs()) + fx.forgetDeliveries() _, err := fx.push() activeTeammates = prev if err == nil || !strings.Contains(err.Error(), "teammate state") { diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index a776dc2..5d6d1c6 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "sync" + "sync/atomic" "time" "github.com/fsnotify/fsnotify" @@ -53,6 +54,11 @@ type Config struct { // Defends against fsnotify event loss on macOS. Defaults to 30s. BaselineTick time.Duration + // PushTimeout bounds each Push call. Defaults to 60s. `source --watch` + // passes the same budget as `source --once` so slow uplinks are not cut + // off early. + PushTimeout time.Duration + // LogLabel prefixes the watcher's stderr lines so they name the command // that owns the loop. Defaults to "agentcookie source --watch". The // cmux-sync local loop passes its own label so logs aren't mislabeled. @@ -89,10 +95,24 @@ func (c Config) baselineTick() time.Duration { return 30 * time.Second } +func (c Config) pushTimeout() time.Duration { + if c.PushTimeout > 0 { + return c.PushTimeout + } + return 60 * time.Second +} + // Watcher runs the watch loop. Construct via New, then call Run. type Watcher struct { cfg Config + // inFlight is set while a Push runs. Triggers that arrive meanwhile are + // dropped instead of queuing behind it: with a multi-minute PushTimeout + // and 30s baseline ticks, queued pushes would otherwise run back to back + // on exactly the slow links the longer budget is for. The next trigger + // after the push finishes sends the current state. + inFlight atomic.Bool + mu sync.Mutex lastPush time.Time lastErr error @@ -145,7 +165,7 @@ func (w *Watcher) Run(ctx context.Context) error { } // Kick off one push at startup so the sink is current from t=0. - go w.runOne(ctx, "startup") + w.start(ctx, "startup", false) debounceTimer := time.NewTimer(time.Hour) debounceTimer.Stop() @@ -190,19 +210,13 @@ func (w *Watcher) Run(ctx context.Context) error { case <-debounceTimer.C: pendingEvent = false - if !w.respectRateCap() { - continue - } - go w.runOne(ctx, "fs-event") + w.start(ctx, "fs-event", true) case <-baselineTicker.C: if pendingEvent { continue } - if !w.respectRateCap() { - continue - } - go w.runOne(ctx, "baseline-tick") + w.start(ctx, "baseline-tick", true) } } } @@ -240,8 +254,26 @@ func (w *Watcher) respectRateCap() bool { return true } +// start launches runOne in a goroutine unless a push is already in flight +// or, when rateCap is set, the rate cap has not elapsed. It reports whether +// a push started. +func (w *Watcher) start(ctx context.Context, reason string, rateCap bool) bool { + if !w.inFlight.CompareAndSwap(false, true) { + return false + } + if rateCap && !w.respectRateCap() { + w.inFlight.Store(false) + return false + } + go func() { + defer w.inFlight.Store(false) + w.runOne(ctx, reason) + }() + return true +} + func (w *Watcher) runOne(ctx context.Context, reason string) { - pushCtx, cancel := context.WithTimeout(ctx, 60*time.Second) + pushCtx, cancel := context.WithTimeout(ctx, w.cfg.pushTimeout()) defer cancel() n, err := w.cfg.Push(pushCtx) w.mu.Lock() diff --git a/internal/watcher/watcher_test.go b/internal/watcher/watcher_test.go index 6f5d986..ae0153e 100644 --- a/internal/watcher/watcher_test.go +++ b/internal/watcher/watcher_test.go @@ -267,3 +267,129 @@ func TestStatsAfterSuccessfulPush(t *testing.T) { t.Errorf("expected ErrorCount == 0, got %d", s.ErrorCount) } } + +func TestConfigPushTimeoutDefault(t *testing.T) { + if got := (Config{}).pushTimeout(); got != 60*time.Second { + t.Errorf("default pushTimeout = %s, want 60s", got) + } + if got := (Config{PushTimeout: 5 * time.Minute}).pushTimeout(); got != 5*time.Minute { + t.Errorf("override pushTimeout = %s, want 5m", got) + } +} + +// pushDeadline runs a watcher with the given PushTimeout and returns how far +// in the future the startup push's context deadline was. +func pushDeadline(t *testing.T, timeout time.Duration) time.Duration { + t.Helper() + dir := t.TempDir() + cookies := filepath.Join(dir, "Cookies") + if err := os.WriteFile(cookies, []byte{}, 0o600); err != nil { + t.Fatal(err) + } + got := make(chan time.Duration, 1) + w, err := New(Config{ + CookiesPath: cookies, + Push: func(ctx context.Context) (int, error) { + dl, ok := ctx.Deadline() + if !ok { + got <- 0 + return 0, nil + } + select { + case got <- time.Until(dl): + default: + } + return 0, nil + }, + PushTimeout: timeout, + BaselineTick: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + // No parent deadline, so the push context's deadline is PushTimeout's. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = w.Run(ctx) }() + select { + case d := <-got: + return d + case <-time.After(time.Second): + t.Fatal("expected startup push within 1s") + return 0 + } +} + +func TestWatcherPushUsesPushTimeout(t *testing.T) { + d := pushDeadline(t, 5*time.Minute) + if d < 4*time.Minute || d > 5*time.Minute { + t.Errorf("push deadline in %s, want about 5m", d) + } +} + +func TestWatcherPushTimeoutDefaultsTo60s(t *testing.T) { + d := pushDeadline(t, 0) + if d < 50*time.Second || d > 60*time.Second { + t.Errorf("push deadline in %s, want about 60s", d) + } +} + +func TestWatcherSkipsTriggersWhilePushInFlight(t *testing.T) { + dir := t.TempDir() + cookies := filepath.Join(dir, "Cookies") + if err := os.WriteFile(cookies, []byte{}, 0o600); err != nil { + t.Fatal(err) + } + release := make(chan struct{}) + var calls, active, maxActive atomic.Int64 + w, err := New(Config{ + CookiesPath: cookies, + Push: func(ctx context.Context) (int, error) { + calls.Add(1) + n := active.Add(1) + defer active.Add(-1) + for { + m := maxActive.Load() + if n <= m || maxActive.CompareAndSwap(m, n) { + break + } + } + select { + case <-release: + case <-ctx.Done(): + } + return 0, nil + }, + Debounce: 10 * time.Millisecond, + MinInterval: time.Millisecond, + BaselineTick: 20 * time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + go func() { _ = w.Run(ctx) }() + + // The startup push blocks. Baseline ticks and file writes keep firing + // while it runs; none of them may start a second push. + for i := range 10 { + _ = os.WriteFile(cookies, []byte{byte(i)}, 0o600) + time.Sleep(30 * time.Millisecond) + } + if got := calls.Load(); got != 1 { + t.Fatalf("Push calls while the first was in flight = %d, want 1", got) + } + + close(release) + deadline := time.Now().Add(2 * time.Second) + for calls.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := calls.Load(); got < 2 { + t.Fatalf("Push calls after release = %d, want a new push once the first finished", got) + } + if got := maxActive.Load(); got != 1 { + t.Errorf("max concurrent Push calls = %d, want 1", got) + } +} From 4afef5d115bdf23883b4c0c9e8012dcfe59713a8 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:01:01 -0700 Subject: [PATCH 2/7] fix(sink): re-inject live CDP cookies every 30s on the sink The source's 30s re-push used to re-inject cookies into a Linux sink's live Chrome as a side effect, which covered browser contexts opened since the last sync and a Chrome that restarted. Now that unchanged pushes are skipped, the sink holds the cookies from its last /sync and re-injects them locally every 30s, applying its current cookie policy. Failures log once and recovery logs once. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- CHANGELOG.md | 6 ++ internal/cli/sink.go | 144 ++++++++++++++++++++++++++- internal/cli/sink_test.go | 204 +++++++++++++++++++++++++++++++++++++- 3 files changed, 350 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4347df6..cb34bfc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## [Unreleased] +### Source push cadence on slow links + +- `source --watch` now gives each push the same time budget as `source --once` (the sync client's per-sink timeout plus 30 seconds) instead of a fixed 60 seconds, so large payloads on slow uplinks stop timing out. A trigger that arrives while a push is still running is skipped instead of queuing behind it; the next baseline tick sends the current state. `cmux-sync` and `agent-sync` keep the 60 second budget. Reported by @Dannyxlm in #149. +- `source --watch` no longer re-sends an unchanged payload every 30 seconds. Each sink is sent the payload only when it changed since that sink last accepted it, or as a heartbeat once 10 minutes have passed. A failed push is always retried with the next tick, and a skipped send counts as neither a push nor a failure in `status`. Changes to a cookie's last-access time alone do not count as a change; the heartbeat carries them. `--verbose` logs each skipped send. Reported by @Dannyxlm in #149. +- A Linux sink with `live_cdp` on now re-injects the cookies from its last `/sync` into the running Chrome every 30 seconds, locally. Browser contexts an agent opened since the last push, and a Chrome that restarted, get cookies within 30 seconds, as they did when the source re-sent every tick. The cookies are held in memory only and pass the sink's current cookie policy on each tick. A failing re-inject is logged once, and again once when it recovers. Pull-mode sinks are unchanged. Reported by @Dannyxlm in #149. + ### Live browser injection keeps your logins - Chrome holds at most about 3300 cookies and evicts the least recently used past that, counting an injected cookie as used when it is set. A large batch (4868 cookies in the live test) silently evicted the first ~1800 it set, which included GitHub's session, so the agent's real browser stayed logged out while the push reported success. Live CDP injection now pushes only the 3000 most recently used cookies, oldest first, so the ones you use are kept. The sink log says how many were pushed and how many stayed sidecar-only. diff --git a/internal/cli/sink.go b/internal/cli/sink.go index 2c08aa1..a2423df 100644 --- a/internal/cli/sink.go +++ b/internal/cli/sink.go @@ -167,7 +167,9 @@ func runSink(cmd *cobra.Command, args []string) error { } key, seqTracker, stateWriter, sinkState, stateMu := rt.key, rt.seq, rt.writer, rt.st, &rt.mu - mux := newSinkMux(cfg, transportSecret, key, seqTracker, stateWriter, sinkState, stateMu) + reinject := newLiveCDPReinjector(cfg) + mux := newSinkMux(cfg, transportSecret, key, seqTracker, stateWriter, sinkState, stateMu, reinject) + startLiveCDPReinject(cmd.Context(), cfg, reinject) srv := httpserver.Configure(&http.Server{Addr: cfg.Listen.Addr, Handler: mux}, httpserver.SinkSync) if sinkDryRun { @@ -307,6 +309,7 @@ func newSinkMux( stateWriter *state.Writer, sinkState *state.SinkState, stateMu *sync.Mutex, + reinject *liveCDPReinjector, ) *http.ServeMux { mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { @@ -323,7 +326,9 @@ func newSinkMux( http.Error(w, "read body: "+err.Error(), http.StatusBadRequest) return } + reinject.beginApply() res := applySealedEnvelope(r.Context(), cfg, transportSecret, key, seqTracker, stateWriter, sinkState, stateMu, sealed) + reinject.endApply(res) if res.Status != http.StatusOK { http.Error(w, res.Reply, res.Status) return @@ -339,6 +344,9 @@ func newSinkMux( type envelopeApplyResult struct { Status int Reply string + // liveCDPCookies is the policy-filtered cookie set a successful apply + // handed to live CDP, for the /sync re-inject ticker. + liveCDPCookies []chrome.Cookie } func applySealedEnvelope( @@ -466,7 +474,7 @@ func applySealedEnvelope( if liveCDPEndpoint == "" { liveCDPEndpoint = livecdp.DefaultCDPEndpoint } - liveCDPContexts, liveCDPErr = livecdp.AttachAndInject(ctx, liveCDPEndpoint, cookies) + liveCDPContexts, liveCDPErr = liveCDPInject(ctx, liveCDPEndpoint, cookies) if liveCDPErr != nil { fmt.Fprintf(os.Stderr, "agentcookie sink: live CDP injection failed (sidecar write succeeded): %v\n", liveCDPErr) } else { @@ -592,7 +600,11 @@ func applySealedEnvelope( okLine += fmt.Sprintf("; live_cdp: no cookies to inject (endpoint=%s)", endpoint) } } - return envelopeApplyResult{Status: http.StatusOK, Reply: okLine} + res := envelopeApplyResult{Status: http.StatusOK, Reply: okLine} + if cfg.LiveCDP.Enabled { + res.liveCDPCookies = cookies + } + return res } func recordSinkReject(sinkState *state.SinkState, stateWriter *state.Writer, stateMu *sync.Mutex, err error) { @@ -1006,3 +1018,129 @@ func isIPBoundLocally(ipStr string) bool { } return false } + +// liveCDPInject injects cookies into the running Chrome at a CDP endpoint. +// Tests replace it so they never dial Chrome. +var liveCDPInject = livecdp.AttachAndInject + +// liveCDPReinjectInterval is how often a live CDP sink re-injects the +// cookies from its last /sync. It matches the source's baseline tick, which +// used to re-push every 30s and so re-injected as a side effect. +const liveCDPReinjectInterval = 30 * time.Second + +// liveCDPReinjector re-injects the cookies from the last successful /sync +// into the running Chrome on a timer (KTD10). The source now skips a sink +// that already holds the current payload and re-sends only every 10 +// minutes, so without this a browser context an agent opened since the last +// /sync, or a Chrome that restarted, would stay logged out until then. The +// cookies are held in memory only; after a sink restart the source's next +// change or heartbeat supplies them. +type liveCDPReinjector struct { + cfg *config.SinkConfig + log io.Writer + + mu sync.Mutex + cookies []chrome.Cookie + applying int + + // failing is the last re-inject outcome, so a persistent failure logs + // once and recovery logs once instead of every tick. Only the ticker + // goroutine touches it. + failing bool +} + +func newLiveCDPReinjector(cfg *config.SinkConfig) *liveCDPReinjector { + return &liveCDPReinjector{cfg: cfg, log: os.Stderr} +} + +// beginApply marks a /sync apply in progress; ticks skip until endApply. +func (r *liveCDPReinjector) beginApply() { + r.mu.Lock() + defer r.mu.Unlock() + r.applying++ +} + +// endApply ends a /sync apply and, when it succeeded with live CDP on, +// holds its cookie set for the next ticks. +func (r *liveCDPReinjector) endApply(res envelopeApplyResult) { + r.mu.Lock() + defer r.mu.Unlock() + r.applying-- + if res.Status == http.StatusOK && r.cfg.LiveCDP.Enabled { + r.cookies = res.liveCDPCookies + } +} + +// tick re-injects the held cookies once. It reports whether it called the +// injector: it skips while no set is held or a /sync apply is running. +func (r *liveCDPReinjector) tick(ctx context.Context) bool { + r.mu.Lock() + cookies, busy := r.cookies, r.applying > 0 + r.mu.Unlock() + if busy || len(cookies) == 0 || !r.cfg.LiveCDP.Enabled { + return false + } + // Re-check the sink's current policy, as /sync would: a domain the + // operator blocked since the last /sync must not keep reaching new + // browser contexts until the source's next heartbeat. + bl, err := loadFreshBlocklist() + if err != nil { + r.report(0, 0, fmt.Errorf("load cookie policy: %w", err)) + return false + } + cookies, _ = protocol.NewBlocklistMatcherForSink(bl).Filter(cookies) + if len(cookies) == 0 { + return false + } + // Bound each attempt so a hung CDP call cannot stop the ticker for good. + injectCtx, cancel := context.WithTimeout(ctx, liveCDPReinjectInterval) + defer cancel() + contexts, err := liveCDPInject(injectCtx, r.endpoint(), cookies) + r.report(len(cookies), contexts, err) + return true +} + +// report logs a re-inject outcome only when it differs from the last one. +func (r *liveCDPReinjector) report(cookies, contexts int, err error) { + switch { + case err != nil && !r.failing: + r.failing = true + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject failed at %s: %v (retrying every %s; logged again on recovery)\n", r.endpoint(), err, liveCDPReinjectInterval) + case err == nil && r.failing: + r.failing = false + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject recovered: %d cookies into %d context(s) at %s\n", cookies, contexts, r.endpoint()) + } +} + +func (r *liveCDPReinjector) endpoint() string { + if r.cfg.LiveCDP.Endpoint != "" { + return r.cfg.LiveCDP.Endpoint + } + return livecdp.DefaultCDPEndpoint +} + +// run calls tick on every value from ticks until ctx ends. +func (r *liveCDPReinjector) run(ctx context.Context, ticks <-chan time.Time) { + for { + select { + case <-ctx.Done(): + return + case <-ticks: + r.tick(ctx) + } + } +} + +// startLiveCDPReinject starts the re-inject ticker when live CDP is on and +// the sink may modify Chrome (not --dry-run). It reports whether it started. +func startLiveCDPReinject(ctx context.Context, cfg *config.SinkConfig, r *liveCDPReinjector) bool { + if !cfg.LiveCDP.Enabled || sinkDryRun { + return false + } + ticker := time.NewTicker(liveCDPReinjectInterval) + go func() { + defer ticker.Stop() + r.run(ctx, ticker.C) + }() + return true +} diff --git a/internal/cli/sink_test.go b/internal/cli/sink_test.go index 3afebbf..1095da3 100644 --- a/internal/cli/sink_test.go +++ b/internal/cli/sink_test.go @@ -16,6 +16,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/mvanhorn/agentcookie/internal/chrome" "github.com/mvanhorn/agentcookie/internal/config" @@ -481,6 +482,7 @@ type sinkHandlerFixture struct { sinkState *state.SinkState stateWriter *state.Writer stateMu *sync.Mutex + reinject *liveCDPReinjector } func newSinkHandlerFixture(t *testing.T, dryRun bool) *sinkHandlerFixture { @@ -505,7 +507,8 @@ func newSinkHandlerFixture(t *testing.T, dryRun bool) *sinkHandlerFixture { stateWriter := state.NewWriter(filepath.Join(t.TempDir(), "sink-state.json")) var stateMu sync.Mutex key := []byte("0123456789abcdef") - mux := newSinkMux(cfg, secret, key, seqTracker, stateWriter, sinkState, &stateMu) + reinject := newLiveCDPReinjector(cfg) + mux := newSinkMux(cfg, secret, key, seqTracker, stateWriter, sinkState, &stateMu, reinject) return &sinkHandlerFixture{ configDir: configDir, @@ -518,6 +521,7 @@ func newSinkHandlerFixture(t *testing.T, dryRun bool) *sinkHandlerFixture { sinkState: sinkState, stateWriter: stateWriter, stateMu: &stateMu, + reinject: reinject, } } @@ -1067,3 +1071,201 @@ func TestAdapterCookieSet_ConcurrentRefusalReadOnce(t *testing.T) { t.Errorf("Safe Storage reads across 16 concurrent batches = %d, want 1", n) } } + +// --- Live CDP local re-inject (KTD10, #149) --- + +type liveCDPInjectRecorder struct { + mu sync.Mutex + calls [][]chrome.Cookie + err error + seen chan struct{} +} + +func (r *liveCDPInjectRecorder) callCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.calls) +} + +func (r *liveCDPInjectRecorder) callAt(i int) []chrome.Cookie { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls[i] +} + +func (r *liveCDPInjectRecorder) setErr(err error) { + r.mu.Lock() + defer r.mu.Unlock() + r.err = err +} + +// stubLiveCDPInject replaces the live CDP injector so no test dials Chrome. +func stubLiveCDPInject(t *testing.T) *liveCDPInjectRecorder { + t.Helper() + rec := &liveCDPInjectRecorder{seen: make(chan struct{}, 16)} + prev := liveCDPInject + liveCDPInject = func(_ context.Context, _ string, cookies []chrome.Cookie) (int, error) { + rec.mu.Lock() + rec.calls = append(rec.calls, append([]chrome.Cookie(nil), cookies...)) + err := rec.err + rec.mu.Unlock() + select { + case rec.seen <- struct{}{}: + default: + } + if err != nil { + return 0, err + } + return 1, nil + } + t.Cleanup(func() { liveCDPInject = prev }) + return rec +} + +// newLiveCDPSinkFixture is a sink handler fixture with live CDP on, a +// sync-all policy on every platform, and a stubbed injector. +func newLiveCDPSinkFixture(t *testing.T) (*sinkHandlerFixture, *liveCDPInjectRecorder, *bytes.Buffer) { + t.Helper() + rec := stubLiveCDPInject(t) + fx := newSinkHandlerFixture(t, false) + fx.cfg.LiveCDP = config.LiveCDPRef{Enabled: true, Endpoint: "http://127.0.0.1:9223"} + writeCLIFile(t, filepath.Join(fx.configDir, "blocklist.yaml"), ` +version: 1 +policy: blocklist +domains: [] +`) + var logBuf bytes.Buffer + fx.reinject.log = &logBuf + return fx, rec, &logBuf +} + +func TestLiveCDPReinjectAfterSync(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + cookies := []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}} + if resp := fx.postSync(1, cookies); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d: %s", resp.Code, resp.Body.String()) + } + if got := rec.callCount(); got != 1 { + t.Fatalf("/sync live CDP injects = %d, want 1", got) + } + + ticks := make(chan time.Time) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + fx.reinject.run(ctx, ticks) + close(done) + }() + <-rec.seen // drain the /sync inject + ticks <- time.Now() + select { + case <-rec.seen: + case <-time.After(2 * time.Second): + t.Fatal("tick did not re-inject") + } + cancel() + <-done + if got := rec.callCount(); got != 2 { + t.Fatalf("injects after one tick = %d, want 2", got) + } + if got := rec.callAt(1); !reflect.DeepEqual(got, cookies) { + t.Fatalf("re-injected %v, want the /sync cookies %v", got, cookies) + } +} + +func TestLiveCDPReinjectSkipsBeforeFirstSync(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + if fx.reinject.tick(context.Background()) { + t.Fatal("tick before any /sync should skip") + } + if got := rec.callCount(); got != 0 { + t.Fatalf("injects before any /sync = %d, want 0", got) + } +} + +func TestLiveCDPReinjectSkipsWhileSyncApplies(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}}); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + fx.reinject.beginApply() + if fx.reinject.tick(context.Background()) { + t.Fatal("tick during a /sync apply should skip") + } + fx.reinject.endApply(envelopeApplyResult{Status: http.StatusConflict}) + if !fx.reinject.tick(context.Background()) { + t.Fatal("tick after the apply finished should re-inject") + } + if got := rec.callCount(); got != 2 { + t.Fatalf("injects = %d, want 2 (/sync + one tick)", got) + } +} + +func TestLiveCDPReinjectNotStartedWhenDisabled(t *testing.T) { + rec := stubLiveCDPInject(t) + fx := newSinkHandlerFixture(t, false) + writeCLIFile(t, filepath.Join(fx.configDir, "blocklist.yaml"), ` +version: 1 +policy: blocklist +domains: [] +`) + if startLiveCDPReinject(context.Background(), fx.cfg, fx.reinject) { + t.Fatal("re-inject ticker started with live_cdp disabled") + } + if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}}); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + if fx.reinject.tick(context.Background()) { + t.Fatal("tick with live_cdp disabled should not inject") + } + if got := rec.callCount(); got != 0 { + t.Fatalf("injects with live_cdp disabled = %d, want 0", got) + } +} + +func TestLiveCDPReinjectLogsFailureOnceThenRecoveryOnce(t *testing.T) { + fx, rec, logBuf := newLiveCDPSinkFixture(t) + if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}}); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + rec.setErr(errors.New("cdp endpoint gone")) + for range 4 { + fx.reinject.tick(context.Background()) + } + if n := strings.Count(logBuf.String(), "cdp endpoint gone"); n != 1 { + t.Fatalf("failure logged %d times over 4 failing ticks, want 1:\n%s", n, logBuf.String()) + } + rec.setErr(nil) + for range 3 { + fx.reinject.tick(context.Background()) + } + if n := strings.Count(logBuf.String(), "recovered"); n != 1 { + t.Fatalf("recovery logged %d times, want 1:\n%s", n, logBuf.String()) + } + if got := rec.callCount(); got != 8 { + t.Fatalf("injects = %d, want 8 (/sync + 7 ticks)", got) + } +} + +func TestLiveCDPReinjectAppliesCurrentSinkPolicy(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + if resp := fx.postSync(1, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + {HostKey: ".blocked.com", Name: "b", Value: "v", Path: "/"}, + }); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + writeCLIFile(t, filepath.Join(fx.configDir, "blocklist.yaml"), ` +version: 1 +policy: blocklist +domains: + - pattern: "%.blocked.com" +`) + if !fx.reinject.tick(context.Background()) { + t.Fatal("tick should re-inject") + } + if got := hostsFromChromeCookies(rec.callAt(1)); !reflect.DeepEqual(got, []string{".example.com"}) { + t.Fatalf("re-injected hosts = %v, want the newly blocked host dropped", got) + } +} From cc7760c01a432839ac89191c71e8162c7eafd41c Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:05:11 -0700 Subject: [PATCH 3/7] refactor(sink): share the live CDP endpoint default and split re-inject logging Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- internal/cli/sink.go | 50 ++++++++++++++++++++++++------------------ internal/cli/source.go | 4 ++-- 2 files changed, 31 insertions(+), 23 deletions(-) diff --git a/internal/cli/sink.go b/internal/cli/sink.go index a2423df..10882f0 100644 --- a/internal/cli/sink.go +++ b/internal/cli/sink.go @@ -470,10 +470,7 @@ func applySealedEnvelope( var liveCDPErr error var liveCDPEndpoint string if cfg.LiveCDP.Enabled && len(cookies) > 0 { - liveCDPEndpoint = cfg.LiveCDP.Endpoint - if liveCDPEndpoint == "" { - liveCDPEndpoint = livecdp.DefaultCDPEndpoint - } + liveCDPEndpoint = liveCDPEndpointFor(cfg) liveCDPContexts, liveCDPErr = liveCDPInject(ctx, liveCDPEndpoint, cookies) if liveCDPErr != nil { fmt.Fprintf(os.Stderr, "agentcookie sink: live CDP injection failed (sidecar write succeeded): %v\n", liveCDPErr) @@ -1025,11 +1022,11 @@ var liveCDPInject = livecdp.AttachAndInject // liveCDPReinjectInterval is how often a live CDP sink re-injects the // cookies from its last /sync. It matches the source's baseline tick, which -// used to re-push every 30s and so re-injected as a side effect. +// re-pushed every 30s and so re-injected as a side effect. const liveCDPReinjectInterval = 30 * time.Second // liveCDPReinjector re-injects the cookies from the last successful /sync -// into the running Chrome on a timer (KTD10). The source now skips a sink +// into the running Chrome on a timer. The source skips a sink // that already holds the current payload and re-sends only every 10 // minutes, so without this a browser context an agent opened since the last // /sync, or a Chrome that restarted, would stay logged out until then. The @@ -1085,7 +1082,7 @@ func (r *liveCDPReinjector) tick(ctx context.Context) bool { // browser contexts until the source's next heartbeat. bl, err := loadFreshBlocklist() if err != nil { - r.report(0, 0, fmt.Errorf("load cookie policy: %w", err)) + r.reportFailure(fmt.Errorf("load cookie policy: %w", err)) return false } cookies, _ = protocol.NewBlocklistMatcherForSink(bl).Filter(cookies) @@ -1095,26 +1092,37 @@ func (r *liveCDPReinjector) tick(ctx context.Context) bool { // Bound each attempt so a hung CDP call cannot stop the ticker for good. injectCtx, cancel := context.WithTimeout(ctx, liveCDPReinjectInterval) defer cancel() - contexts, err := liveCDPInject(injectCtx, r.endpoint(), cookies) - r.report(len(cookies), contexts, err) + contexts, err := liveCDPInject(injectCtx, liveCDPEndpointFor(r.cfg), cookies) + if err != nil { + r.reportFailure(err) + } else { + r.reportRecovery(len(cookies), contexts) + } return true } -// report logs a re-inject outcome only when it differs from the last one. -func (r *liveCDPReinjector) report(cookies, contexts int, err error) { - switch { - case err != nil && !r.failing: - r.failing = true - fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject failed at %s: %v (retrying every %s; logged again on recovery)\n", r.endpoint(), err, liveCDPReinjectInterval) - case err == nil && r.failing: - r.failing = false - fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject recovered: %d cookies into %d context(s) at %s\n", cookies, contexts, r.endpoint()) +// reportFailure logs a failed re-inject once, until a recovery. +func (r *liveCDPReinjector) reportFailure(err error) { + if r.failing { + return + } + r.failing = true + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject failed at %s: %v (retrying every %s; logged again on recovery)\n", liveCDPEndpointFor(r.cfg), err, liveCDPReinjectInterval) +} + +// reportRecovery logs the first successful re-inject after a failure. +func (r *liveCDPReinjector) reportRecovery(cookies, contexts int) { + if !r.failing { + return } + r.failing = false + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject recovered: %d cookies into %d context(s) at %s\n", cookies, contexts, liveCDPEndpointFor(r.cfg)) } -func (r *liveCDPReinjector) endpoint() string { - if r.cfg.LiveCDP.Endpoint != "" { - return r.cfg.LiveCDP.Endpoint +// liveCDPEndpointFor is the sink's live CDP endpoint, or the default. +func liveCDPEndpointFor(cfg *config.SinkConfig) string { + if cfg.LiveCDP.Endpoint != "" { + return cfg.LiveCDP.Endpoint } return livecdp.DefaultCDPEndpoint } diff --git a/internal/cli/source.go b/internal/cli/source.go index cbf2a57..16bb736 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -101,7 +101,7 @@ const ( // sinkHeartbeat is how long an unchanged payload goes without being re-sent // to a sink that already received it. doctor only warns after 24h without a // push, so 10 minutes keeps status fresh while an idle source stops sending -// the same envelope every 30s (#149). +// the same envelope every 30s. const sinkHeartbeat = 10 * time.Minute // sinkDelivery is the last payload a sink accepted: the envelope hash (see @@ -387,7 +387,7 @@ func runSource(cmd *cobra.Command, args []string) error { // sourcePushBudget bounds one whole source push cycle, in --once and in // --watch alike: SyncClient's per-sink timeout plus a small slack for // envelope packing. Pre-v0.12 --once was hardcoded at 60s, which was tight -// even for v0.10-shape payloads, and --watch kept that 60s until #149. The +// even for v0.10-shape payloads. The // inner HTTP request also bounds itself; this outer cancel is the belt to the // request's suspenders. // From a1960067bf1476b69dafe60b7fb0a8bbd5387277 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:21:36 -0700 Subject: [PATCH 4/7] fix(source,sink): address review findings on push dedup and re-inject - Forget a sink's remembered delivery when a POST is sent and fails, so a later payload is never skipped on an older acknowledgement. - Stamp each delivery when its POST succeeds, so a slow POST does not bring the next heartbeat forward. - Serialize the live CDP re-inject tick with /sync applies so a newer /sync set always lands after an in-flight tick. - Use liveCDPEndpointFor everywhere, and move the re-injector and the dedup log into sink_reinject.go and source_dedup.go. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- internal/cli/doctor.go | 6 +- internal/cli/sink.go | 147 +----------------------------- internal/cli/sink_reinject.go | 162 ++++++++++++++++++++++++++++++++++ internal/cli/sink_test.go | 75 ++++++++++++++++ internal/cli/source.go | 89 ++----------------- internal/cli/source_dedup.go | 100 +++++++++++++++++++++ internal/cli/source_test.go | 76 ++++++++++++++++ 7 files changed, 422 insertions(+), 233 deletions(-) create mode 100644 internal/cli/sink_reinject.go create mode 100644 internal/cli/source_dedup.go diff --git a/internal/cli/doctor.go b/internal/cli/doctor.go index f9cc6bc..bacd791 100644 --- a/internal/cli/doctor.go +++ b/internal/cli/doctor.go @@ -23,7 +23,6 @@ import ( "github.com/mvanhorn/agentcookie/internal/config" "github.com/mvanhorn/agentcookie/internal/keystore" "github.com/mvanhorn/agentcookie/internal/launchd" - "github.com/mvanhorn/agentcookie/internal/livecdp" "github.com/mvanhorn/agentcookie/internal/secretsbus" "github.com/mvanhorn/agentcookie/internal/sinkpush" "github.com/mvanhorn/agentcookie/internal/state" @@ -1431,10 +1430,7 @@ func checkLiveCDPEndpointWith(sinkCfg *config.SinkConfig, probe func(endpoint st } } - endpoint := sinkCfg.LiveCDP.Endpoint - if endpoint == "" { - endpoint = livecdp.DefaultCDPEndpoint - } + endpoint := liveCDPEndpointFor(sinkCfg) if err := probe(endpoint); err == nil { return Check{ diff --git a/internal/cli/sink.go b/internal/cli/sink.go index 10882f0..a0540b6 100644 --- a/internal/cli/sink.go +++ b/internal/cli/sink.go @@ -136,10 +136,7 @@ func runSink(cmd *cobra.Command, args []string) error { // Log live CDP mode at startup (Linux sink's primary injection path). if cfg.LiveCDP.Enabled { - endpoint := cfg.LiveCDP.Endpoint - if endpoint == "" { - endpoint = livecdp.DefaultCDPEndpoint - } + endpoint := liveCDPEndpointFor(cfg) fmt.Fprintf(os.Stderr, "agentcookie sink: live_cdp enabled; will inject into running Chrome at %s\n", endpoint) } transportSecret, err := resolveTransportSecret(common.ConfigDir, cfg.Peer.Hostname, cfg.Security.SharedSecret) @@ -584,10 +581,7 @@ func applySealedEnvelope( if cfg.LiveCDP.Enabled { endpoint := liveCDPEndpoint if endpoint == "" { - endpoint = cfg.LiveCDP.Endpoint - if endpoint == "" { - endpoint = livecdp.DefaultCDPEndpoint - } + endpoint = liveCDPEndpointFor(cfg) } if liveCDPErr != nil { okLine += fmt.Sprintf("; live_cdp: FAILED at %s: %v", endpoint, liveCDPErr) @@ -1015,140 +1009,3 @@ func isIPBoundLocally(ipStr string) bool { } return false } - -// liveCDPInject injects cookies into the running Chrome at a CDP endpoint. -// Tests replace it so they never dial Chrome. -var liveCDPInject = livecdp.AttachAndInject - -// liveCDPReinjectInterval is how often a live CDP sink re-injects the -// cookies from its last /sync. It matches the source's baseline tick, which -// re-pushed every 30s and so re-injected as a side effect. -const liveCDPReinjectInterval = 30 * time.Second - -// liveCDPReinjector re-injects the cookies from the last successful /sync -// into the running Chrome on a timer. The source skips a sink -// that already holds the current payload and re-sends only every 10 -// minutes, so without this a browser context an agent opened since the last -// /sync, or a Chrome that restarted, would stay logged out until then. The -// cookies are held in memory only; after a sink restart the source's next -// change or heartbeat supplies them. -type liveCDPReinjector struct { - cfg *config.SinkConfig - log io.Writer - - mu sync.Mutex - cookies []chrome.Cookie - applying int - - // failing is the last re-inject outcome, so a persistent failure logs - // once and recovery logs once instead of every tick. Only the ticker - // goroutine touches it. - failing bool -} - -func newLiveCDPReinjector(cfg *config.SinkConfig) *liveCDPReinjector { - return &liveCDPReinjector{cfg: cfg, log: os.Stderr} -} - -// beginApply marks a /sync apply in progress; ticks skip until endApply. -func (r *liveCDPReinjector) beginApply() { - r.mu.Lock() - defer r.mu.Unlock() - r.applying++ -} - -// endApply ends a /sync apply and, when it succeeded with live CDP on, -// holds its cookie set for the next ticks. -func (r *liveCDPReinjector) endApply(res envelopeApplyResult) { - r.mu.Lock() - defer r.mu.Unlock() - r.applying-- - if res.Status == http.StatusOK && r.cfg.LiveCDP.Enabled { - r.cookies = res.liveCDPCookies - } -} - -// tick re-injects the held cookies once. It reports whether it called the -// injector: it skips while no set is held or a /sync apply is running. -func (r *liveCDPReinjector) tick(ctx context.Context) bool { - r.mu.Lock() - cookies, busy := r.cookies, r.applying > 0 - r.mu.Unlock() - if busy || len(cookies) == 0 || !r.cfg.LiveCDP.Enabled { - return false - } - // Re-check the sink's current policy, as /sync would: a domain the - // operator blocked since the last /sync must not keep reaching new - // browser contexts until the source's next heartbeat. - bl, err := loadFreshBlocklist() - if err != nil { - r.reportFailure(fmt.Errorf("load cookie policy: %w", err)) - return false - } - cookies, _ = protocol.NewBlocklistMatcherForSink(bl).Filter(cookies) - if len(cookies) == 0 { - return false - } - // Bound each attempt so a hung CDP call cannot stop the ticker for good. - injectCtx, cancel := context.WithTimeout(ctx, liveCDPReinjectInterval) - defer cancel() - contexts, err := liveCDPInject(injectCtx, liveCDPEndpointFor(r.cfg), cookies) - if err != nil { - r.reportFailure(err) - } else { - r.reportRecovery(len(cookies), contexts) - } - return true -} - -// reportFailure logs a failed re-inject once, until a recovery. -func (r *liveCDPReinjector) reportFailure(err error) { - if r.failing { - return - } - r.failing = true - fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject failed at %s: %v (retrying every %s; logged again on recovery)\n", liveCDPEndpointFor(r.cfg), err, liveCDPReinjectInterval) -} - -// reportRecovery logs the first successful re-inject after a failure. -func (r *liveCDPReinjector) reportRecovery(cookies, contexts int) { - if !r.failing { - return - } - r.failing = false - fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject recovered: %d cookies into %d context(s) at %s\n", cookies, contexts, liveCDPEndpointFor(r.cfg)) -} - -// liveCDPEndpointFor is the sink's live CDP endpoint, or the default. -func liveCDPEndpointFor(cfg *config.SinkConfig) string { - if cfg.LiveCDP.Endpoint != "" { - return cfg.LiveCDP.Endpoint - } - return livecdp.DefaultCDPEndpoint -} - -// run calls tick on every value from ticks until ctx ends. -func (r *liveCDPReinjector) run(ctx context.Context, ticks <-chan time.Time) { - for { - select { - case <-ctx.Done(): - return - case <-ticks: - r.tick(ctx) - } - } -} - -// startLiveCDPReinject starts the re-inject ticker when live CDP is on and -// the sink may modify Chrome (not --dry-run). It reports whether it started. -func startLiveCDPReinject(ctx context.Context, cfg *config.SinkConfig, r *liveCDPReinjector) bool { - if !cfg.LiveCDP.Enabled || sinkDryRun { - return false - } - ticker := time.NewTicker(liveCDPReinjectInterval) - go func() { - defer ticker.Stop() - r.run(ctx, ticker.C) - }() - return true -} diff --git a/internal/cli/sink_reinject.go b/internal/cli/sink_reinject.go new file mode 100644 index 0000000..1f57c6e --- /dev/null +++ b/internal/cli/sink_reinject.go @@ -0,0 +1,162 @@ +package cli + +import ( + "context" + "fmt" + "io" + "net/http" + "os" + "sync" + "time" + + "github.com/mvanhorn/agentcookie/internal/chrome" + "github.com/mvanhorn/agentcookie/internal/config" + "github.com/mvanhorn/agentcookie/internal/livecdp" + "github.com/mvanhorn/agentcookie/internal/protocol" +) + +// liveCDPInject injects cookies into the running Chrome at a CDP endpoint. +// Tests replace it so they never dial Chrome. +var liveCDPInject = livecdp.AttachAndInject + +// liveCDPReinjectInterval is how often a live CDP sink re-injects the +// cookies from its last /sync. It matches the source's baseline tick, which +// re-pushed every 30s and so re-injected as a side effect. +const liveCDPReinjectInterval = 30 * time.Second + +// liveCDPReinjector re-injects the cookies from the last successful /sync +// into the running Chrome on a timer. The source skips a sink +// that already holds the current payload and re-sends only every 10 +// minutes, so without this a browser context an agent opened since the last +// /sync, or a Chrome that restarted, would stay logged out until then. The +// cookies are held in memory only; after a sink restart the source's next +// change or heartbeat supplies them. +type liveCDPReinjector struct { + cfg *config.SinkConfig + log io.Writer + + mu sync.Mutex + cookies []chrome.Cookie + applying int + + // injectMu covers a tick's snapshot and inject, so a /sync that starts + // mid-tick waits for it and its newer cookies are injected last. + injectMu sync.Mutex + + // failing is the last re-inject outcome, so a persistent failure logs + // once and recovery logs once instead of every tick. Only the ticker + // goroutine touches it. + failing bool +} + +func newLiveCDPReinjector(cfg *config.SinkConfig) *liveCDPReinjector { + return &liveCDPReinjector{cfg: cfg, log: os.Stderr} +} + +// beginApply marks a /sync apply in progress; ticks skip until endApply. +// It first waits for an in-flight tick to finish injecting. +func (r *liveCDPReinjector) beginApply() { + r.injectMu.Lock() + defer r.injectMu.Unlock() + r.mu.Lock() + defer r.mu.Unlock() + r.applying++ +} + +// endApply ends a /sync apply and, when it succeeded with live CDP on, +// holds its cookie set for the next ticks. +func (r *liveCDPReinjector) endApply(res envelopeApplyResult) { + r.mu.Lock() + defer r.mu.Unlock() + r.applying-- + if res.Status == http.StatusOK && r.cfg.LiveCDP.Enabled { + r.cookies = res.liveCDPCookies + } +} + +// tick re-injects the held cookies once. It reports whether it called the +// injector: it skips while no set is held or a /sync apply is running. +func (r *liveCDPReinjector) tick(ctx context.Context) bool { + r.injectMu.Lock() + defer r.injectMu.Unlock() + r.mu.Lock() + cookies, busy := r.cookies, r.applying > 0 + r.mu.Unlock() + if busy || len(cookies) == 0 || !r.cfg.LiveCDP.Enabled { + return false + } + // Re-check the sink's current policy, as /sync would: a domain the + // operator blocked since the last /sync must not keep reaching new + // browser contexts until the source's next heartbeat. + bl, err := loadFreshBlocklist() + if err != nil { + r.reportFailure(fmt.Errorf("load cookie policy: %w", err)) + return false + } + cookies, _ = protocol.NewBlocklistMatcherForSink(bl).Filter(cookies) + if len(cookies) == 0 { + return false + } + // Bound each attempt so a hung CDP call cannot stop the ticker for good. + injectCtx, cancel := context.WithTimeout(ctx, liveCDPReinjectInterval) + defer cancel() + contexts, err := liveCDPInject(injectCtx, liveCDPEndpointFor(r.cfg), cookies) + if err != nil { + r.reportFailure(err) + } else { + r.reportRecovery(len(cookies), contexts) + } + return true +} + +// reportFailure logs a failed re-inject once, until a recovery. +func (r *liveCDPReinjector) reportFailure(err error) { + if r.failing { + return + } + r.failing = true + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject failed at %s: %v (retrying every %s; logged again on recovery)\n", liveCDPEndpointFor(r.cfg), err, liveCDPReinjectInterval) +} + +// reportRecovery logs the first successful re-inject after a failure. +func (r *liveCDPReinjector) reportRecovery(cookies, contexts int) { + if !r.failing { + return + } + r.failing = false + fmt.Fprintf(r.log, "agentcookie sink: live CDP re-inject recovered: %d cookies into %d context(s) at %s\n", cookies, contexts, liveCDPEndpointFor(r.cfg)) +} + +// liveCDPEndpointFor is the sink's live CDP endpoint, or the default. +func liveCDPEndpointFor(cfg *config.SinkConfig) string { + if cfg.LiveCDP.Endpoint != "" { + return cfg.LiveCDP.Endpoint + } + return livecdp.DefaultCDPEndpoint +} + +// run calls tick on every value from ticks until ctx ends. +func (r *liveCDPReinjector) run(ctx context.Context, ticks <-chan time.Time) { + for { + select { + case <-ctx.Done(): + return + case <-ticks: + r.tick(ctx) + } + } +} + +// startLiveCDPReinject starts the re-inject ticker when live CDP is on and +// the sink may modify Chrome (not --dry-run). It reports whether it started. +func startLiveCDPReinject(ctx context.Context, cfg *config.SinkConfig, r *liveCDPReinjector) bool { + if !cfg.LiveCDP.Enabled || sinkDryRun { + return false + } + ticker := time.NewTicker(liveCDPReinjectInterval) + go func() { + defer ticker.Stop() + r.run(ctx, ticker.C) + }() + return true +} diff --git a/internal/cli/sink_test.go b/internal/cli/sink_test.go index 1095da3..83f01cb 100644 --- a/internal/cli/sink_test.go +++ b/internal/cli/sink_test.go @@ -1269,3 +1269,78 @@ domains: t.Fatalf("re-injected hosts = %v, want the newly blocked host dropped", got) } } + +func TestLiveCDPReinjectSyncLandsAfterInFlightTick(t *testing.T) { + fx, _, _ := newLiveCDPSinkFixture(t) + if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "old", Path: "/"}}); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + + // The tick's inject of the old set blocks until released. + var mu sync.Mutex + var injected []string + tickEntered := make(chan struct{}) + releaseTick := make(chan struct{}) + newInjected := make(chan struct{}, 1) + liveCDPInject = func(_ context.Context, _ string, cookies []chrome.Cookie) (int, error) { + v := cookies[0].Value + if v == "old" { + close(tickEntered) + <-releaseTick + } + mu.Lock() + injected = append(injected, v) + mu.Unlock() + if v == "new" { + newInjected <- struct{}{} + } + return 1, nil + } + + tickDone := make(chan struct{}) + go func() { + fx.reinject.tick(context.Background()) + close(tickDone) + }() + <-tickEntered + + syncDone := make(chan int, 1) + go func() { + syncDone <- fx.postSync(2, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "new", Path: "/"}}).Code + }() + // The /sync must not inject while the tick is mid-inject. + select { + case <-newInjected: + case <-time.After(300 * time.Millisecond): + } + close(releaseTick) + <-tickDone + if code := <-syncDone; code != http.StatusOK { + t.Fatalf("second /sync status = %d", code) + } + + mu.Lock() + defer mu.Unlock() + if want := []string{"old", "new"}; !reflect.DeepEqual(injected, want) { + t.Fatalf("inject order = %v, want %v: the newer /sync set must land last", injected, want) + } +} + +func TestLiveCDPReinjectNotStartedInDryRun(t *testing.T) { + rec := stubLiveCDPInject(t) + fx := newSinkHandlerFixture(t, true) + fx.cfg.LiveCDP = config.LiveCDPRef{Enabled: true, Endpoint: "http://127.0.0.1:9223"} + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + if startLiveCDPReinject(ctx, fx.cfg, fx.reinject) { + t.Fatal("re-inject ticker started in --dry-run") + } + if got := rec.callCount(); got != 0 { + t.Fatalf("injects in --dry-run = %d, want 0", got) + } + + sinkDryRun = false + if !startLiveCDPReinject(ctx, fx.cfg, fx.reinject) { + t.Fatal("re-inject ticker not started with live_cdp on and --dry-run off") + } +} diff --git a/internal/cli/source.go b/internal/cli/source.go index 16bb736..4f621bf 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -3,7 +3,6 @@ package cli import ( "bytes" "context" - "crypto/sha256" "encoding/json" "errors" "fmt" @@ -98,87 +97,6 @@ const ( sinkBackoffMax = 30 * time.Minute ) -// sinkHeartbeat is how long an unchanged payload goes without being re-sent -// to a sink that already received it. doctor only warns after 24h without a -// push, so 10 minutes keeps status fresh while an idle source stops sending -// the same envelope every 30s. -const sinkHeartbeat = 10 * time.Minute - -// sinkDelivery is the last payload a sink accepted: the envelope hash (see -// envelopeHash) and when it was sent. -type sinkDelivery struct { - hash [sha256.Size]byte - at time.Time -} - -// sinkDeliveryLog remembers, per sink (peer + URL), the last successful -// delivery of this process. It lives in memory only: a restarted source -// sends everything once, which is what --once and a fresh daemon want. -type sinkDeliveryLog struct { - mu sync.Mutex - m map[string]sinkDelivery -} - -func newSinkDeliveryLog() *sinkDeliveryLog { - return &sinkDeliveryLog{m: map[string]sinkDelivery{}} -} - -// sinkDeliveries is the source process's delivery log. Tests replace it so -// one test's deliveries do not suppress another's. -var sinkDeliveries = newSinkDeliveryLog() - -func sinkDeliveryKey(peer, rawURL string) string { - return peer + "\x00" + rawURL -} - -// unchanged reports whether the sink already received hash less than -// sinkHeartbeat before now, and if so when. -func (l *sinkDeliveryLog) unchanged(peer, rawURL string, hash [sha256.Size]byte, now time.Time) (time.Time, bool) { - l.mu.Lock() - defer l.mu.Unlock() - d, ok := l.m[sinkDeliveryKey(peer, rawURL)] - if !ok || d.hash != hash || now.Sub(d.at) >= sinkHeartbeat { - return time.Time{}, false - } - return d.at, true -} - -// record notes a successful delivery. Only a sink that accepted the POST is -// recorded, so a failed push is always re-sent. -func (l *sinkDeliveryLog) record(peer, rawURL string, hash [sha256.Size]byte, now time.Time) { - l.mu.Lock() - defer l.mu.Unlock() - l.m[sinkDeliveryKey(peer, rawURL)] = sinkDelivery{hash: hash, at: now} -} - -// envelopeHash identifies an envelope's content for the unchanged check. -// Sequence is zeroed: every real POST needs a fresh one for the sink's -// replay tracker, so it must not make identical content look new. -// -// Cookie LastAccessUTC is zeroed too, in the hash only (the sent payload -// keeps it). Chrome bumps last_access_utc whenever a request sends a cookie, -// throttled to once a minute per cookie (CookieMonster's 60s access-update -// threshold), and its SQLite store commits pending writes every 30s. With -// any tab open, some access time changes between most 30s baseline ticks, -// which would defeat the check. The sink only uses access times to order -// live CDP injection, and the heartbeat carries fresh ones every 10 minutes. -func envelopeHash(env protocol.SyncEnvelope) ([sha256.Size]byte, error) { - env.Sequence = 0 - cookies := make([]chrome.Cookie, len(env.Cookies)) - for i, c := range env.Cookies { - c.LastAccessUTC = 0 - cookies[i] = c - } - env.Cookies = cookies - h := sha256.New() - if err := json.NewEncoder(h).Encode(env); err != nil { - return [sha256.Size]byte{}, err - } - var sum [sha256.Size]byte - copy(sum[:], h.Sum(nil)) - return sum, nil -} - // errSinkAsleep marks a sink whose Tailscale peer is offline. The source // skips it without dialing and records it as asleep, not as a failure. var errSinkAsleep = errors.New("sink peer is offline") @@ -929,9 +847,15 @@ func pushOnce( r.Asleep = true case postErr != nil: r.Err = postErr + // A failed request may have partly changed the sink, so an + // older acknowledgement no longer proves what it holds. + sinkDeliveries.forget(sink.Peer, sink.URL) default: r.Count = len(all) *reply = body + // Stamp the delivery when the sink accepted it, so a slow + // POST does not bring the next heartbeat forward. + sinkDeliveries.record(sink.Peer, sink.URL, hash, sourceNow()) } }(&results[i], &replies[i], sink) } @@ -961,7 +885,6 @@ func pushOnce( humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] push failed: %v", r.URL, r.Err)) default: delivered++ - sinkDeliveries.record(r.Peer, r.URL, hash, now) rep["posted"] = true humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] posted %d cookies, sink replied: %s%s", r.URL, r.Count, replies[i], dbscNote(dbsc))) } diff --git a/internal/cli/source_dedup.go b/internal/cli/source_dedup.go new file mode 100644 index 0000000..41fca31 --- /dev/null +++ b/internal/cli/source_dedup.go @@ -0,0 +1,100 @@ +package cli + +import ( + "crypto/sha256" + "encoding/json" + "sync" + "time" + + "github.com/mvanhorn/agentcookie/internal/chrome" + "github.com/mvanhorn/agentcookie/internal/protocol" +) + +// sinkHeartbeat is how long an unchanged payload goes without being re-sent +// to a sink that already received it. doctor only warns after 24h without a +// push, so 10 minutes keeps status fresh while an idle source stops sending +// the same envelope every 30s. +const sinkHeartbeat = 10 * time.Minute + +// sinkDelivery is the last payload a sink accepted: the envelope hash (see +// envelopeHash) and when the sink accepted it. +type sinkDelivery struct { + hash [sha256.Size]byte + at time.Time +} + +// sinkDeliveryLog remembers, per sink (peer + URL), the last successful +// delivery of this process. It lives in memory only: a restarted source +// sends everything once, which is what --once and a fresh daemon want. +type sinkDeliveryLog struct { + mu sync.Mutex + m map[string]sinkDelivery +} + +func newSinkDeliveryLog() *sinkDeliveryLog { + return &sinkDeliveryLog{m: map[string]sinkDelivery{}} +} + +// sinkDeliveries is the source process's delivery log. Tests replace it so +// one test's deliveries do not suppress another's. +var sinkDeliveries = newSinkDeliveryLog() + +func sinkDeliveryKey(peer, rawURL string) string { + return peer + "\x00" + rawURL +} + +// unchanged reports whether the sink already received hash less than +// sinkHeartbeat before now, and if so when. +func (l *sinkDeliveryLog) unchanged(peer, rawURL string, hash [sha256.Size]byte, now time.Time) (time.Time, bool) { + l.mu.Lock() + defer l.mu.Unlock() + d, ok := l.m[sinkDeliveryKey(peer, rawURL)] + if !ok || d.hash != hash || now.Sub(d.at) >= sinkHeartbeat { + return time.Time{}, false + } + return d.at, true +} + +// record notes a successful delivery. Only a sink that accepted the POST is +// recorded, so a failed push is always re-sent. +func (l *sinkDeliveryLog) record(peer, rawURL string, hash [sha256.Size]byte, now time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + l.m[sinkDeliveryKey(peer, rawURL)] = sinkDelivery{hash: hash, at: now} +} + +// forget drops a sink's delivery, so its next push is sent whatever the +// payload. A failed POST calls it. +func (l *sinkDeliveryLog) forget(peer, rawURL string) { + l.mu.Lock() + defer l.mu.Unlock() + delete(l.m, sinkDeliveryKey(peer, rawURL)) +} + +// envelopeHash identifies an envelope's content for the unchanged check. +// Sequence is zeroed: every real POST needs a fresh one for the sink's +// replay tracker, so it must not make identical content look new. +// +// Cookie LastAccessUTC is zeroed too, in the hash only (the sent payload +// keeps it). Chrome bumps last_access_utc whenever a request sends a cookie, +// throttled to once a minute per cookie (CookieMonster's 60s access-update +// threshold), and its SQLite store commits pending writes every 30s. With +// any tab open, some access time changes between most 30s baseline ticks, +// which would defeat the check. The sink only uses access times to order +// live CDP injection, and the heartbeat carries fresh ones every 10 minutes. +func envelopeHash(env protocol.SyncEnvelope) ([sha256.Size]byte, error) { + env.Sequence = 0 + cookies := make([]chrome.Cookie, len(env.Cookies)) + for i, c := range env.Cookies { + c.LastAccessUTC = 0 + cookies[i] = c + } + env.Cookies = cookies + h := sha256.New() + if err := json.NewEncoder(h).Encode(env); err != nil { + return [sha256.Size]byte{}, err + } + var sum [sha256.Size]byte + copy(sum[:], h.Sum(nil)) + return sum, nil +} diff --git a/internal/cli/source_test.go b/internal/cli/source_test.go index 7dc93c4..c92ec46 100644 --- a/internal/cli/source_test.go +++ b/internal/cli/source_test.go @@ -442,6 +442,9 @@ type sourceCapture struct { attempts map[string]int // arrivals records when each URL's most recent POST reached the transport. arrivals map[string]time.Time + // onPost, when set, runs as an accepted POST completes, standing in for + // a sink that takes a while to reply. + onPost func() } func newSourceCapture(t *testing.T, secret string) *sourceCapture { @@ -490,7 +493,11 @@ func (c *sourceCapture) RoundTrip(req *http.Request) (*http.Response, error) { c.batches = append(c.batches, append([]chrome.Cookie(nil), envelope.Cookies...)) c.urls = append(c.urls, req.URL.String()) c.envelopes = append(c.envelopes, envelope) + onPost := c.onPost c.mu.Unlock() + if onPost != nil { + onPost() + } return &http.Response{ StatusCode: http.StatusOK, @@ -1407,3 +1414,72 @@ func TestSourcePushUnchangedSkipLeavesFailureStateAlone(t *testing.T) { t.Errorf("source TotalFailures = %d, want %d", fx.srcState.TotalFailures, srcBefore) } } + +func TestSourcePushFailedPostForgetsEarlierDelivery(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "a", Path: "/"}, + }) + sinkURL := fx.cfg.Sink.URL + now := stubSourceClock(t, time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)) + + if _, err := fx.push(); err != nil { + t.Fatalf("push A: %v", err) + } + // Payload B fails: the sink may have partly applied it. + seedSourceCookiesDB(t, fx.cfg.Chrome.DBPath, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "b", Path: "/"}, + }, fx.key) + fx.capture.failURL = sinkURL + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err == nil { + t.Fatal("push B should fail") + } + fx.capture.failURL = "" + // Back to payload A, past the backoff window and inside A's heartbeat. + seedSourceCookiesDB(t, fx.cfg.Chrome.DBPath, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "a", Path: "/"}, + }, fx.key) + *now = now.Add(30 * time.Second) + if _, err := fx.push(); err != nil { + t.Fatalf("push A again: %v", err) + } + if got := fx.capture.attemptsTo(sinkURL); got != 3 { + t.Fatalf("POST attempts = %d, want 3: A must be re-sent after a failed POST", got) + } + if got := fx.batchCount(); got != 2 { + t.Fatalf("delivered batches = %d, want 2", got) + } +} + +func TestSourcePushHeartbeatCountsFromPostCompletion(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + sinkURL := fx.cfg.Sink.URL + start := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC) + now := stubSourceClock(t, start) + + // The first POST takes 2 minutes to complete. + fx.capture.onPost = func() { *now = now.Add(2 * time.Minute) } + if _, err := fx.push(); err != nil { + t.Fatalf("first push: %v", err) + } + fx.capture.onPost = nil + + // 10 minutes after the push started, but only 8 after the sink accepted. + *now = start.Add(sinkHeartbeat) + if _, err := fx.push(); err != nil { + t.Fatalf("second push: %v", err) + } + if got := fx.capture.attemptsTo(sinkURL); got != 1 { + t.Fatalf("POSTs = %d, want 1: heartbeat is due 10 minutes after the POST completed", got) + } + + *now = start.Add(2*time.Minute + sinkHeartbeat) + if _, err := fx.push(); err != nil { + t.Fatalf("heartbeat push: %v", err) + } + if got := fx.capture.attemptsTo(sinkURL); got != 2 { + t.Fatalf("POSTs after heartbeat = %d, want 2", got) + } +} From a0c1a4d3bbdb38e8dde8440f85374a81f9c0478d Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:24:13 -0700 Subject: [PATCH 5/7] refactor(source): a hash failure only disables the unchanged check Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- internal/cli/source.go | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/internal/cli/source.go b/internal/cli/source.go index 4f621bf..5d44554 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -770,10 +770,9 @@ func pushOnce( return nil, dbsc, fmt.Errorf("marshal envelope: %w", err) } pullPayloadCache.StoreIfCurrent(pullGen, payload) - hash, err := envelopeHash(envelope) - if err != nil { - return nil, dbsc, fmt.Errorf("hash envelope: %w", err) - } + // A hash failure only turns off the unchanged check for this cycle, so + // every sink is sent the payload. + hash, hashErr := envelopeHash(envelope) // Fan out: read and filtering above happened once; only sealing and // transport repeat per sink. Each sink is sealed with its own key and // POSTed concurrently under its own timeout, so a dead sink cannot hold @@ -819,7 +818,7 @@ func pushOnce( continue } } - if at, ok := sinkDeliveries.unchanged(sink.Peer, sink.URL, hash, now); ok { + if at, ok := sinkDeliveries.unchanged(sink.Peer, sink.URL, hash, now); hashErr == nil && ok { results[i].Unchanged = true results[i].DeliveredAt = at continue @@ -855,7 +854,9 @@ func pushOnce( *reply = body // Stamp the delivery when the sink accepted it, so a slow // POST does not bring the next heartbeat forward. - sinkDeliveries.record(sink.Peer, sink.URL, hash, sourceNow()) + if hashErr == nil { + sinkDeliveries.record(sink.Peer, sink.URL, hash, sourceNow()) + } } }(&results[i], &replies[i], sink) } From 2b544d71ee3697da23cbf1cb0025d0723c5c9df6 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:33:03 -0700 Subject: [PATCH 6/7] fix(sink): keep the newest re-inject set and record tick outcomes in status Addresses review feedback on #154. An older /sync that finishes after a newer one from the same source no longer replaces the held cookie set, and each re-inject tick that attempts an inject records its outcome in the sink state, so status reflects the latest live CDP result. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- internal/cli/sink.go | 46 +++++++++++------ internal/cli/sink_reinject.go | 43 ++++++++++++++-- internal/cli/sink_test.go | 97 ++++++++++++++++++++++++++++++++++- 3 files changed, 163 insertions(+), 23 deletions(-) diff --git a/internal/cli/sink.go b/internal/cli/sink.go index a0540b6..ea74663 100644 --- a/internal/cli/sink.go +++ b/internal/cli/sink.go @@ -164,7 +164,7 @@ func runSink(cmd *cobra.Command, args []string) error { } key, seqTracker, stateWriter, sinkState, stateMu := rt.key, rt.seq, rt.writer, rt.st, &rt.mu - reinject := newLiveCDPReinjector(cfg) + reinject := newLiveCDPReinjector(cfg, stateWriter, sinkState, stateMu) mux := newSinkMux(cfg, transportSecret, key, seqTracker, stateWriter, sinkState, stateMu, reinject) startLiveCDPReinject(cmd.Context(), cfg, reinject) @@ -344,6 +344,10 @@ type envelopeApplyResult struct { // liveCDPCookies is the policy-filtered cookie set a successful apply // handed to live CDP, for the /sync re-inject ticker. liveCDPCookies []chrome.Cookie + // sourceHostname and sequence identify the accepted envelope, so the + // ticker never replaces a newer held set with an older one. + sourceHostname string + sequence int64 } func applySealedEnvelope( @@ -549,22 +553,7 @@ func applySealedEnvelope( sinkState.TotalDropped += dropped if cfg.LiveCDP.Enabled && len(cookies) > 0 { sinkState.LastWriteMode = writeMode + "+livecdp" - if sinkState.LiveCDP == nil { - sinkState.LiveCDP = &state.LiveCDPState{ - Enabled: true, - Endpoint: liveCDPEndpoint, - } - } - sinkState.LiveCDP.LastInjectAt = time.Now().UTC() - sinkState.LiveCDP.LastCookies = len(cookies) - sinkState.LiveCDP.LastContexts = liveCDPContexts - if liveCDPErr != nil { - sinkState.LiveCDP.LastError = liveCDPErr.Error() - sinkState.LiveCDP.TotalFailures++ - } else { - sinkState.LiveCDP.LastError = "" - sinkState.LiveCDP.TotalInjects++ - } + recordLiveCDPInject(sinkState, liveCDPEndpoint, len(cookies), liveCDPContexts, liveCDPErr) } if len(adapterResults) > 0 { sinkState.LastAdapterResults = toStateAdapterResults(adapterResults) @@ -594,10 +583,33 @@ func applySealedEnvelope( res := envelopeApplyResult{Status: http.StatusOK, Reply: okLine} if cfg.LiveCDP.Enabled { res.liveCDPCookies = cookies + res.sourceHostname = envelope.SourceHostname + res.sequence = envelope.Sequence } return res } +// recordLiveCDPInject records one live CDP inject attempt in sinkState for +// `agentcookie status`. The caller holds the state mutex. +func recordLiveCDPInject(sinkState *state.SinkState, endpoint string, cookies, contexts int, err error) { + if sinkState.LiveCDP == nil { + sinkState.LiveCDP = &state.LiveCDPState{ + Enabled: true, + Endpoint: endpoint, + } + } + sinkState.LiveCDP.LastInjectAt = time.Now().UTC() + sinkState.LiveCDP.LastCookies = cookies + sinkState.LiveCDP.LastContexts = contexts + if err != nil { + sinkState.LiveCDP.LastError = err.Error() + sinkState.LiveCDP.TotalFailures++ + } else { + sinkState.LiveCDP.LastError = "" + sinkState.LiveCDP.TotalInjects++ + } +} + func recordSinkReject(sinkState *state.SinkState, stateWriter *state.Writer, stateMu *sync.Mutex, err error) { if sinkState == nil { return diff --git a/internal/cli/sink_reinject.go b/internal/cli/sink_reinject.go index 1f57c6e..5b40784 100644 --- a/internal/cli/sink_reinject.go +++ b/internal/cli/sink_reinject.go @@ -13,6 +13,7 @@ import ( "github.com/mvanhorn/agentcookie/internal/config" "github.com/mvanhorn/agentcookie/internal/livecdp" "github.com/mvanhorn/agentcookie/internal/protocol" + "github.com/mvanhorn/agentcookie/internal/state" ) // liveCDPInject injects cookies into the running Chrome at a CDP endpoint. @@ -35,9 +36,18 @@ type liveCDPReinjector struct { cfg *config.SinkConfig log io.Writer + // stateWriter, sinkState, and stateMu are the /sync path's status + // state, so each tick's outcome reaches `agentcookie status`. + stateWriter *state.Writer + sinkState *state.SinkState + stateMu *sync.Mutex + mu sync.Mutex cookies []chrome.Cookie applying int + // heldSource and heldSeq identify the envelope cookies came from. + heldSource string + heldSeq int64 // injectMu covers a tick's snapshot and inject, so a /sync that starts // mid-tick waits for it and its newer cookies are injected last. @@ -49,8 +59,8 @@ type liveCDPReinjector struct { failing bool } -func newLiveCDPReinjector(cfg *config.SinkConfig) *liveCDPReinjector { - return &liveCDPReinjector{cfg: cfg, log: os.Stderr} +func newLiveCDPReinjector(cfg *config.SinkConfig, stateWriter *state.Writer, sinkState *state.SinkState, stateMu *sync.Mutex) *liveCDPReinjector { + return &liveCDPReinjector{cfg: cfg, log: os.Stderr, stateWriter: stateWriter, sinkState: sinkState, stateMu: stateMu} } // beginApply marks a /sync apply in progress; ticks skip until endApply. @@ -64,14 +74,22 @@ func (r *liveCDPReinjector) beginApply() { } // endApply ends a /sync apply and, when it succeeded with live CDP on, -// holds its cookie set for the next ticks. +// holds its cookie set for the next ticks. Concurrent applies can finish +// out of order, so an older envelope from the held source never replaces +// a newer one; a different source replaces it (last writer wins). func (r *liveCDPReinjector) endApply(res envelopeApplyResult) { r.mu.Lock() defer r.mu.Unlock() r.applying-- - if res.Status == http.StatusOK && r.cfg.LiveCDP.Enabled { - r.cookies = res.liveCDPCookies + if res.Status != http.StatusOK || !r.cfg.LiveCDP.Enabled { + return } + if r.cookies != nil && res.sourceHostname == r.heldSource && res.sequence < r.heldSeq { + return + } + r.cookies = res.liveCDPCookies + r.heldSource = res.sourceHostname + r.heldSeq = res.sequence } // tick re-injects the held cookies once. It reports whether it called the @@ -101,6 +119,7 @@ func (r *liveCDPReinjector) tick(ctx context.Context) bool { injectCtx, cancel := context.WithTimeout(ctx, liveCDPReinjectInterval) defer cancel() contexts, err := liveCDPInject(injectCtx, liveCDPEndpointFor(r.cfg), cookies) + r.recordState(len(cookies), contexts, err) if err != nil { r.reportFailure(err) } else { @@ -109,6 +128,20 @@ func (r *liveCDPReinjector) tick(ctx context.Context) bool { return true } +// recordState records a re-inject attempt's outcome in the sink status +// state, as the /sync path does. +func (r *liveCDPReinjector) recordState(cookies, contexts int, err error) { + if r.sinkState == nil { + return + } + r.stateMu.Lock() + defer r.stateMu.Unlock() + recordLiveCDPInject(r.sinkState, liveCDPEndpointFor(r.cfg), cookies, contexts, err) + if r.stateWriter != nil { + _ = r.stateWriter.Save(r.sinkState) + } +} + // reportFailure logs a failed re-inject once, until a recovery. func (r *liveCDPReinjector) reportFailure(err error) { if r.failing { diff --git a/internal/cli/sink_test.go b/internal/cli/sink_test.go index 83f01cb..ff67043 100644 --- a/internal/cli/sink_test.go +++ b/internal/cli/sink_test.go @@ -507,7 +507,7 @@ func newSinkHandlerFixture(t *testing.T, dryRun bool) *sinkHandlerFixture { stateWriter := state.NewWriter(filepath.Join(t.TempDir(), "sink-state.json")) var stateMu sync.Mutex key := []byte("0123456789abcdef") - reinject := newLiveCDPReinjector(cfg) + reinject := newLiveCDPReinjector(cfg, stateWriter, sinkState, &stateMu) mux := newSinkMux(cfg, secret, key, seqTracker, stateWriter, sinkState, &stateMu, reinject) return &sinkHandlerFixture{ @@ -1344,3 +1344,98 @@ func TestLiveCDPReinjectNotStartedInDryRun(t *testing.T) { t.Fatal("re-inject ticker not started with live_cdp on and --dry-run off") } } + +// sealedEnvelope seals one envelope from the fixture's test source. +func (f *sinkHandlerFixture) sealedEnvelope(t *testing.T, seq int64, cookies []chrome.Cookie) []byte { + t.Helper() + payload, err := json.Marshal(protocol.SyncEnvelope{ + ProtocolVersion: protocol.Version, + SourceHostname: "source-test", + Sequence: seq, + Cookies: cookies, + }) + if err != nil { + t.Fatal(err) + } + sealed, err := transport.SealWithSecret(payload, f.secret) + if err != nil { + t.Fatal(err) + } + return sealed +} + +func TestLiveCDPReinjectKeepsNewerSetWhenOlderApplyFinishesLast(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + // A is accepted before B, but B's request finishes first. + fx.reinject.beginApply() + fx.reinject.beginApply() + resA := fx.applySealed(fx.sealedEnvelope(t, 1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "A", Path: "/"}})) + resB := fx.applySealed(fx.sealedEnvelope(t, 2, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "B", Path: "/"}})) + if resA.Status != http.StatusOK || resB.Status != http.StatusOK { + t.Fatalf("apply status A=%d B=%d, want 200", resA.Status, resB.Status) + } + fx.reinject.endApply(resB) + fx.reinject.endApply(resA) + + if !fx.reinject.tick(context.Background()) { + t.Fatal("tick should re-inject") + } + if got := rec.callAt(rec.callCount() - 1)[0].Value; got != "B" { + t.Fatalf("tick re-injected %q, want the newer B set", got) + } +} + +func TestLiveCDPReinjectTickFailureRecordedInState(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}}); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + rec.setErr(errors.New("cdp endpoint gone")) + if !fx.reinject.tick(context.Background()) { + t.Fatal("tick should attempt a re-inject") + } + fx.stateMu.Lock() + defer fx.stateMu.Unlock() + cdp := fx.sinkState.LiveCDP + if cdp == nil || !strings.Contains(cdp.LastError, "cdp endpoint gone") { + t.Fatalf("LiveCDP state after failing tick = %+v, want LastError set", cdp) + } + if cdp.TotalFailures != 1 { + t.Fatalf("TotalFailures = %d, want 1", cdp.TotalFailures) + } +} + +func TestLiveCDPReinjectTickSuccessClearsStateError(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + rec.setErr(errors.New("cdp endpoint gone")) + if resp := fx.postSync(1, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + {HostKey: ".example.com", Name: "t", Value: "v", Path: "/"}, + }); resp.Code != http.StatusOK { + t.Fatalf("/sync status = %d", resp.Code) + } + fx.stateMu.Lock() + before := *fx.sinkState.LiveCDP + fx.stateMu.Unlock() + if before.LastError == "" { + t.Fatal("failed /sync inject should set LastError") + } + + rec.setErr(nil) + time.Sleep(time.Millisecond) + if !fx.reinject.tick(context.Background()) { + t.Fatal("tick should re-inject") + } + fx.stateMu.Lock() + defer fx.stateMu.Unlock() + cdp := fx.sinkState.LiveCDP + if cdp.LastError != "" { + t.Fatalf("LastError after succeeding tick = %q, want cleared", cdp.LastError) + } + if !cdp.LastInjectAt.After(before.LastInjectAt) { + t.Fatalf("LastInjectAt = %v, want after %v", cdp.LastInjectAt, before.LastInjectAt) + } + if cdp.LastCookies != 2 || cdp.LastContexts != 1 || cdp.TotalInjects != before.TotalInjects+1 { + t.Fatalf("counts after tick = %+v, want 2 cookies, 1 context, one more inject", cdp) + } +} From bcd6f98fd2d3a43326dad1f76c74a3e01338102a Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:38:13 -0700 Subject: [PATCH 7/7] fix(sink): an older apply cannot replace a newer empty re-inject set Addresses review feedback on #154. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LgVWtkXgDGWMh6L95d8piN --- internal/cli/sink_reinject.go | 8 ++++++-- internal/cli/sink_test.go | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/internal/cli/sink_reinject.go b/internal/cli/sink_reinject.go index 5b40784..07cfb0b 100644 --- a/internal/cli/sink_reinject.go +++ b/internal/cli/sink_reinject.go @@ -45,9 +45,12 @@ type liveCDPReinjector struct { mu sync.Mutex cookies []chrome.Cookie applying int - // heldSource and heldSeq identify the envelope cookies came from. + // heldSource and heldSeq identify the envelope cookies came from. held + // is set once any apply was kept, including one with no cookies, so an + // older apply cannot replace a newer empty set. heldSource string heldSeq int64 + held bool // injectMu covers a tick's snapshot and inject, so a /sync that starts // mid-tick waits for it and its newer cookies are injected last. @@ -84,12 +87,13 @@ func (r *liveCDPReinjector) endApply(res envelopeApplyResult) { if res.Status != http.StatusOK || !r.cfg.LiveCDP.Enabled { return } - if r.cookies != nil && res.sourceHostname == r.heldSource && res.sequence < r.heldSeq { + if r.held && res.sourceHostname == r.heldSource && res.sequence < r.heldSeq { return } r.cookies = res.liveCDPCookies r.heldSource = res.sourceHostname r.heldSeq = res.sequence + r.held = true } // tick re-injects the held cookies once. It reports whether it called the diff --git a/internal/cli/sink_test.go b/internal/cli/sink_test.go index ff67043..57f8e5a 100644 --- a/internal/cli/sink_test.go +++ b/internal/cli/sink_test.go @@ -1385,6 +1385,26 @@ func TestLiveCDPReinjectKeepsNewerSetWhenOlderApplyFinishesLast(t *testing.T) { } } +// A newer /sync whose cookie set is empty must also win over an older apply +// that finishes after it. +func TestLiveCDPReinjectKeepsNewerEmptySetWhenOlderApplyFinishesLast(t *testing.T) { + fx, rec, _ := newLiveCDPSinkFixture(t) + fx.reinject.beginApply() + fx.reinject.beginApply() + resA := fx.applySealed(fx.sealedEnvelope(t, 1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "A", Path: "/"}})) + resB := fx.applySealed(fx.sealedEnvelope(t, 2, nil)) + if resA.Status != http.StatusOK || resB.Status != http.StatusOK { + t.Fatalf("apply status A=%d B=%d, want 200", resA.Status, resB.Status) + } + before := rec.callCount() + fx.reinject.endApply(resB) + fx.reinject.endApply(resA) + + if fx.reinject.tick(context.Background()) { + t.Fatalf("tick re-injected the older A set over the newer empty set (%d calls)", rec.callCount()-before) + } +} + func TestLiveCDPReinjectTickFailureRecordedInState(t *testing.T) { fx, rec, _ := newLiveCDPSinkFixture(t) if resp := fx.postSync(1, []chrome.Cookie{{HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}}); resp.Code != http.StatusOK {