From 3d5db3b6cfba97440835e4b8f1c98839c0bba1d1 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:15:50 -0700 Subject: [PATCH 1/7] feat(config): add multi-sink fan-out list with legacy synthesis Add SinkTarget and SourceConfig.Sinks, plus ResolvedSinks() which returns the sinks list when present and otherwise synthesizes a one-element list from the legacy scalar sink/peer fields, so every pre-multi-sink source.yaml keeps working unchanged. Generalize LoadSource validation to be per-sink (each needs a URL and either a peer key or the legacy shared secret) while preserving the legacy single-sink error behavior. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- internal/config/config.go | 64 ++++++++++++++++++++---- internal/config/config_test.go | 89 ++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 9 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 8ec9124..5562aab 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,11 +19,19 @@ import ( // legacy Security.SharedSecret field is kept for backwards compat with v0 // configs that predate pairing. type SourceConfig struct { - Sink SinkRef `yaml:"sink" json:"sink"` - Chrome ChromeRef `yaml:"chrome" json:"chrome"` - Browser BrowserRef `yaml:"browser,omitempty" json:"browser,omitempty"` - Peer PeerRef `yaml:"peer,omitempty" json:"peer,omitempty"` - Security SecurityRef `yaml:"security,omitempty" json:"security,omitempty"` + // Sinks is the multi-sink fan-out list: each entry carries its own + // URL and peer (key). When non-empty it is authoritative and the + // legacy scalar Sink/Peer fields below are ignored. When empty, a + // legacy single-sink config is synthesized into a one-element list + // from Sink+Peer (see ResolvedSinks), so every pre-multi-sink + // source.yaml keeps working with no migration. omitempty keeps a + // legacy-only config from emitting an empty sinks: key. + Sinks []SinkTarget `yaml:"sinks,omitempty" json:"sinks,omitempty"` + Sink SinkRef `yaml:"sink,omitempty" json:"sink,omitempty"` + Chrome ChromeRef `yaml:"chrome" json:"chrome"` + Browser BrowserRef `yaml:"browser,omitempty" json:"browser,omitempty"` + Peer PeerRef `yaml:"peer,omitempty" json:"peer,omitempty"` + Security SecurityRef `yaml:"security,omitempty" json:"security,omitempty"` // Cmux configures the same-machine local loop: `agentcookie cmux-sync` // reads this machine's Chrome and injects into this machine's cmux // browser. Independent of the sink/peer push path; absent = loop off. @@ -119,6 +127,33 @@ type SinkRef struct { URL string `yaml:"url" json:"url"` } +// SinkTarget is one entry in the multi-sink fan-out list: a sink URL plus +// the peer hostname naming its key under keys/. A push seals the payload +// once per SinkTarget with that peer's key and POSTs it to URL. Peer may +// be empty only for a legacy single-sink config synthesized from the +// scalar Sink/Peer fields, in which case the transport falls back to the +// legacy Security.SharedSecret (see ResolvedSinks and LoadSource). +type SinkTarget struct { + URL string `yaml:"url" json:"url"` + Peer string `yaml:"peer,omitempty" json:"peer,omitempty"` +} + +// ResolvedSinks returns the effective fan-out list. Sinks wins when +// present; otherwise a legacy scalar Sink.URL is synthesized into a +// one-element list carrying the scalar Peer.Hostname. When neither is set +// it returns nil (no delivery), so a mis-written config fails visibly at +// the caller rather than POSTing to an empty URL. Every consumer resolves +// sinks through this method so the legacy fallback lives in one place. +func (c *SourceConfig) ResolvedSinks() []SinkTarget { + if len(c.Sinks) > 0 { + return c.Sinks + } + if c.Sink.URL != "" { + return []SinkTarget{{URL: c.Sink.URL, Peer: c.Peer.Hostname}} + } + return nil +} + type ListenRef struct { Addr string `yaml:"addr" json:"addr"` } @@ -167,11 +202,22 @@ func LoadSource(dir string) (*SourceConfig, error) { if err := loadYAML(path, &cfg); err != nil { return nil, err } - if cfg.Sink.URL == "" { - return nil, fmt.Errorf("%s: sink.url is required", path) + sinks := cfg.ResolvedSinks() + if len(sinks) == 0 { + return nil, fmt.Errorf("%s: at least one sink is required (set sink.url, or a sinks: list)", path) } - if cfg.Peer.Hostname == "" && cfg.Security.SharedSecret == "" { - return nil, fmt.Errorf("%s: either peer.hostname (paired key) or security.shared_secret (legacy) is required", path) + for i, s := range sinks { + if s.URL == "" { + return nil, fmt.Errorf("%s: sinks[%d].url is required", path, i) + } + // Each sink needs a credential: its own peer key, or the legacy + // shared secret when the sink carries no peer (the synthesized + // legacy single-sink case). A sink with no peer and no shared + // secret has no way to seal, so reject it rather than silently + // producing an unsealable push. + if s.Peer == "" && cfg.Security.SharedSecret == "" { + return nil, fmt.Errorf("%s: sink %q needs either a peer (paired key) or security.shared_secret (legacy)", path, s.URL) + } } if err := validateSharedSecret(path, cfg.Security.SharedSecret); err != nil { return nil, err diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ed2a820..7de6e13 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -735,3 +735,92 @@ func writeFile(t *testing.T, dir, name, content string) { t.Fatalf("write %s: %v", name, err) } } + +func TestLoadSourceMultiSinkResolvesAllEntries(t *testing.T) { + dir := t.TempDir() + writeFile(t, dir, "source.yaml", ` +sinks: + - url: http://a.test:9999/sync + peer: alpha + - url: http://b.test:9999/sync + peer: bravo +chrome: + db_path: ~/Library/Application Support/Google/Chrome/Default/Cookies +`) + cfg, err := LoadSource(dir) + if err != nil { + t.Fatalf("LoadSource: %v", err) + } + got := cfg.ResolvedSinks() + if len(got) != 2 { + t.Fatalf("expected 2 resolved sinks, got %d", len(got)) + } + if got[0].URL != "http://a.test:9999/sync" || got[0].Peer != "alpha" { + t.Errorf("sink[0] wrong: %+v", got[0]) + } + if got[1].URL != "http://b.test:9999/sync" || got[1].Peer != "bravo" { + t.Errorf("sink[1] wrong: %+v", got[1]) + } +} + +func TestResolvedSinksSynthesizesLegacyScalar(t *testing.T) { + cfg := &SourceConfig{ + Sink: SinkRef{URL: "http://legacy.test:9999/sync"}, + Peer: PeerRef{Hostname: "legacy-peer"}, + } + got := cfg.ResolvedSinks() + if len(got) != 1 { + t.Fatalf("expected 1 synthesized sink, got %d", len(got)) + } + if got[0].URL != "http://legacy.test:9999/sync" || got[0].Peer != "legacy-peer" { + t.Errorf("synthesized sink wrong: %+v", got[0]) + } +} + +func TestResolvedSinksPrefersSinksOverLegacyScalar(t *testing.T) { + cfg := &SourceConfig{ + Sinks: []SinkTarget{{URL: "http://new.test:9999/sync", Peer: "new-peer"}}, + Sink: SinkRef{URL: "http://legacy.test:9999/sync"}, + Peer: PeerRef{Hostname: "legacy-peer"}, + } + got := cfg.ResolvedSinks() + if len(got) != 1 || got[0].Peer != "new-peer" { + t.Fatalf("expected sinks list to win, got %+v", got) + } +} + +func TestResolvedSinksEmptyWhenNeitherSet(t *testing.T) { + cfg := &SourceConfig{} + if got := cfg.ResolvedSinks(); got != nil { + t.Fatalf("expected nil for no sinks, got %+v", got) + } +} + +func TestLoadSourceMultiSinkRequiresPerSinkPeerOrSharedSecret(t *testing.T) { + dir := t.TempDir() + // A sinks entry with no peer and no legacy shared secret has no way to seal. + writeFile(t, dir, "source.yaml", ` +sinks: + - url: http://a.test:9999/sync +chrome: + db_path: ~/Library/Application Support/Google/Chrome/Default/Cookies +`) + if _, err := LoadSource(dir); err == nil { + t.Fatal("expected error for a sink with no peer and no shared secret, got nil") + } +} + +func TestLoadSourceMultiSinkSharedSecretCoversPeerlessSink(t *testing.T) { + dir := t.TempDir() + writeFile(t, dir, "source.yaml", ` +sinks: + - url: http://a.test:9999/sync +security: + shared_secret: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa +chrome: + db_path: ~/Library/Application Support/Google/Chrome/Default/Cookies +`) + if _, err := LoadSource(dir); err != nil { + t.Fatalf("shared secret should cover a peerless sink: %v", err) + } +} From 81155fe75b0c468cc6c0f70221944470121f9079 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:24:33 -0700 Subject: [PATCH 2/7] feat(source): fan out cookie and secret sync to multiple sinks Read and filter cookies once, then seal and POST per sink. Resolve each sink's transport secret inside the fan-out loop so a missing per-sink key isolates that sink instead of aborting delivery to all, and never silently downgrades a peer'd sink to the fleet-wide shared secret. Per-sink failures are isolated and reported (errors.Join keeps each chain, so ErrAmbiguousPeer still matches); a partial or total sink failure is a non-zero exit. Scale the --once outer deadline by sink count so a slow first sink cannot starve later healthy ones. Track per-sink push state (SourceState.Sinks, keyed by peer hostname, URL refreshed in place) alongside the retained cross-sink aggregate so old state files and single-sink configs keep working unchanged. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- internal/cli/source.go | 249 +++++++++++++++++++++++++++-------- internal/cli/source_test.go | 131 +++++++++++++++++- internal/state/state.go | 44 +++++++ internal/state/state_test.go | 42 ++++++ 4 files changed, 406 insertions(+), 60 deletions(-) diff --git a/internal/cli/source.go b/internal/cli/source.go index e3dcbec..e8dba95 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -9,6 +9,7 @@ import ( "io" "net/http" "os" + "strings" "time" "github.com/spf13/cobra" @@ -17,6 +18,7 @@ import ( "github.com/mvanhorn/agentcookie/internal/chromedirsync" "github.com/mvanhorn/agentcookie/internal/cli/httpserver" "github.com/mvanhorn/agentcookie/internal/config" + "github.com/mvanhorn/agentcookie/internal/keystore" "github.com/mvanhorn/agentcookie/internal/pairing" "github.com/mvanhorn/agentcookie/internal/protocol" "github.com/mvanhorn/agentcookie/internal/secretsbus" @@ -113,22 +115,27 @@ func runSource(cmd *cobra.Command, args []string) error { if err != nil { return err } - secret, err := resolveTransportSecret(common.ConfigDir, cfg.Peer.Hostname, cfg.Security.SharedSecret) - if err != nil { - return err - } + // Per-sink transport secrets are resolved inside the fan-out loop + // (pushOnce), not here: resolving all secrets up front would let one + // missing per-sink key abort delivery to every sink, defeating the + // per-sink failure isolation. A missing key isolates that one sink. + sinks := cfg.ResolvedSinks() // State writer for `agentcookie status` to read. home, _ := os.UserHomeDir() stateWriter := state.NewWriter(state.SourcePath(home)) - srcState := &state.SourceState{Role: "source", SinkURL: cfg.Sink.URL} + legacySinkURL := "" + if len(sinks) > 0 { + legacySinkURL = sinks[0].URL + } + srcState := &state.SourceState{Role: "source", SinkURL: legacySinkURL} // --skip-dbsc-suspect is also honored via env var so a LaunchAgent can // opt in without a flag edit. skipDBSC := sourceSkipDBSC || os.Getenv("AGENTCOOKIE_SKIP_DBSC_SUSPECT") == "1" push := func(ctx context.Context) (int, error) { - return pushWithFreshBlocklist(ctx, cfg, key, secret, sourceDryRun, sourceVerbose, skipDBSC, srcState, stateWriter) + return pushWithFreshBlocklist(ctx, cfg, key, sourceDryRun, sourceVerbose, skipDBSC, srcState, stateWriter) } if sourceOnce { @@ -137,7 +144,18 @@ func runSource(cmd *cobra.Command, args []string) error { // 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. - ctx, cancel := context.WithTimeout(cmd.Context(), httpserver.Defaults(httpserver.SyncClient).ClientTimeout+30*time.Second) + // + // Scale the outer budget by sink count: fan-out POSTs run + // sequentially, so a first sink that runs to its full per-sink + // timeout must not exhaust the shared deadline and starve later + // healthy sinks (which would defeat per-sink isolation). One + // per-sink ClientTimeout per sink, plus slack. + nSinks := len(sinks) + if nSinks < 1 { + nSinks = 1 + } + perSink := httpserver.Defaults(httpserver.SyncClient).ClientTimeout + ctx, cancel := context.WithTimeout(cmd.Context(), time.Duration(nSinks)*perSink+30*time.Second) defer cancel() _, err := push(ctx) return err @@ -160,7 +178,7 @@ func runSource(cmd *cobra.Command, args []string) error { if err != nil { return fmt.Errorf("init watcher: %w", err) } - fmt.Fprintf(os.Stderr, "agentcookie source --watch: adapter=%s watching %s, sink=%s\n", sourceBrowser.Name, cfg.Chrome.DBPath, cfg.Sink.URL) + fmt.Fprintf(os.Stderr, "agentcookie source --watch: adapter=%s watching %s, sinks=%s\n", sourceBrowser.Name, cfg.Chrome.DBPath, sinkURLList(sinks)) // v0.13: also watch ~/.agentcookie/secrets/ so a write to a per-CLI // secrets.env triggers the same push pipeline as a Chrome cookie @@ -203,7 +221,6 @@ func pushWithFreshBlocklist( ctx context.Context, cfg *config.SourceConfig, key []byte, - secret string, dryRun bool, verbose bool, skipDBSC bool, @@ -213,32 +230,81 @@ func pushWithFreshBlocklist( blocklist, err := loadFreshBlocklist() var dbsc dbscSummary if err != nil { - recordSourcePushResult(srcState, stateWriter, 0, dbsc, err) + recordSourcePushResult(srcState, stateWriter, nil, dbsc, err) return 0, err } - n, dbsc, err := pushOnce(ctx, cfg, blocklist, key, secret, dryRun, verbose, skipDBSC) - recordSourcePushResult(srcState, stateWriter, n, dbsc, err) - return n, err + results, dbsc, err := pushOnce(ctx, cfg, blocklist, key, dryRun, verbose, skipDBSC) + recordSourcePushResult(srcState, stateWriter, results, dbsc, err) + if err != nil { + return 0, err + } + // Aggregate per-sink outcomes: partial or total sink failure is a push + // failure for exit-status purposes (KTD3), even though healthy sinks + // still received the payload. errors.Join preserves each sink error's + // chain so errors.Is (e.g. tsclient.ErrAmbiguousPeer) still matches. + posted, sinkErrs := aggregateSinkResults(results) + if len(sinkErrs) > 0 { + return posted, fmt.Errorf("push failed for %d of %d sink(s): %w", len(sinkErrs), len(results), errors.Join(sinkErrs...)) + } + return posted, nil +} + +// aggregateSinkResults returns the cookie count of the most-delivered sink +// (all sinks receive the same set) and the errors of any that failed, each +// prefixed with its sink label so the joined message names the sink. +func aggregateSinkResults(results []sinkResult) (posted int, sinkErrs []error) { + for _, r := range results { + if r.Err != nil { + label := r.URL + if r.Peer != "" { + label = r.Peer + " (" + r.URL + ")" + } + sinkErrs = append(sinkErrs, fmt.Errorf("%s: %w", label, r.Err)) + continue + } + if r.Count > posted { + posted = r.Count + } + } + return posted, sinkErrs } func recordSourcePushResult( srcState *state.SourceState, stateWriter *state.Writer, - n int, + results []sinkResult, dbsc dbscSummary, err error, ) { if srcState == nil { return } + now := time.Now().UTC() + // A read/envelope-phase failure (err != nil) is recorded once at the + // top level; no per-sink attempt happened. if err != nil { srcState.TotalFailures++ srcState.LastError = err.Error() - srcState.LastErrorAt = time.Now().UTC() - } else { - srcState.TotalPushes++ - srcState.LastPushCount = n - srcState.LastPush = time.Now().UTC() + srcState.LastErrorAt = now + } + for _, r := range results { + ps := srcState.SinkFor(r.Peer, r.URL) + if r.Err != nil { + ps.TotalFailures++ + ps.LastError = r.Err.Error() + ps.LastErrorAt = now + srcState.TotalFailures++ + srcState.LastError = r.Err.Error() + srcState.LastErrorAt = now + } else { + ps.TotalPushes++ + ps.LastPushCount = r.Count + ps.LastPush = now + ps.LastError = "" + srcState.TotalPushes++ + srcState.LastPushCount = r.Count + srcState.LastPush = now + } } srcState.LastDBSCWarned = dbsc.warned srcState.LastDBSCSkipped = dbsc.skipped @@ -248,6 +314,23 @@ func recordSourcePushResult( } } +// sinkResult is one sink's outcome from a single fan-out push. +type sinkResult struct { + Peer string + URL string + Count int // cookies posted to this sink (0 on error) + Err error +} + +// sinkURLList renders sink URLs for a log line. +func sinkURLList(sinks []config.SinkTarget) string { + urls := make([]string, 0, len(sinks)) + for _, s := range sinks { + urls = append(urls, s.URL) + } + return strings.Join(urls, ", ") +} + // pushOnce performs one read+filter+push cycle. Returns the number of cookies // successfully posted (0 on dry-run or error). // @@ -259,11 +342,10 @@ func pushOnce( cfg *config.SourceConfig, blocklist *config.Blocklist, key []byte, - secret string, dryRun bool, verbose bool, skipDBSC bool, -) (int, dbscSummary, error) { +) ([]sinkResult, dbscSummary, error) { var dbsc dbscSummary // Shared read pipeline (decrypt -> cookie policy -> DBSC). See @@ -271,7 +353,7 @@ func pushOnce( // both use it so they filter identically. all, st, err := readFilteredCookies(cfg.Chrome.DBPath, blocklist, key, skipDBSC, time.Now().UTC()) if err != nil { - return 0, dbsc, err + return nil, dbsc, err } totalRead := st.totalRead totalDropped := st.totalDropped @@ -333,7 +415,7 @@ func pushOnce( if dryRun || (len(all) == 0 && secretsCLICount == 0) { _ = emit(result, fmt.Sprintf("agentcookie source: %d cookies after cookie policy (%s), %d secrets clis (dry-run=%v)%s\n", len(all), blocklist.CookiePolicySummary(), secretsCLICount, dryRun, dbscNote(dbsc))) - return 0, dbsc, nil + return nil, dbsc, nil } // v0.7: pack Local Storage and IndexedDB alongside cookies from the @@ -347,7 +429,7 @@ func pushOnce( // the rest of this push are reading from the configured browser). sourceBrowser, err := chrome.LookupBrowser(cfg.Browser.Name) if err != nil { - return 0, dbsc, err + return nil, dbsc, err } var lsTarball []byte var idbTarball []byte @@ -385,28 +467,92 @@ func pushOnce( } payload, err := json.Marshal(envelope) if err != nil { - return 0, dbsc, fmt.Errorf("marshal envelope: %w", err) + return nil, dbsc, fmt.Errorf("marshal 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 independently. A per-sink failure (missing key, seal error, + // URL-resolution, or POST) is isolated to that sink and recorded; the + // loop continues so the other sinks still receive the payload. + sinks := cfg.ResolvedSinks() + results := make([]sinkResult, 0, len(sinks)) + sinkReports := make([]map[string]any, 0, len(sinks)) + var humanLines []string + for _, sink := range sinks { + r := sinkResult{Peer: sink.Peer, URL: sink.URL} + secret, secErr := resolveSinkSecret(common.ConfigDir, sink, cfg.Security.SharedSecret) + if secErr != nil { + r.Err = secErr + } else if sealed, sealErr := transport.SealWithSecret(payload, secret); sealErr != nil { + r.Err = fmt.Errorf("seal payload: %w", sealErr) + } else if reply, postErr := postToSink(ctx, sink.URL, sealed, verbose); postErr != nil { + r.Err = postErr + } else { + r.Count = len(all) + humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] posted %d cookies, sink replied: %s%s", sink.URL, len(all), reply, dbscNote(dbsc))) + } + if r.Err != nil { + humanLines = append(humanLines, fmt.Sprintf("agentcookie source: [%s] push failed: %v", sink.URL, r.Err)) + } + results = append(results, r) + rep := map[string]any{"url": sink.URL, "peer": sink.Peer, "posted": r.Err == nil} + if r.Err != nil { + rep["error"] = r.Err.Error() + } + sinkReports = append(sinkReports, rep) } - sealed, err := transport.SealWithSecret(payload, secret) - if err != nil { - return 0, dbsc, fmt.Errorf("seal payload: %w", err) - } - - // Resolve sink URL hostname to IP via Tailscale if needed. This allows - // sink.url to use MagicDNS hostnames (e.g., http://grok-bot:9999/sync) - // instead of frozen 100.x IPs that break after Tailscale re-auth. - // - // Fail-closed errors (ErrAmbiguousPeer) abort the push — falling back to - // the hostname URL would hand selection to MagicDNS and undo fail-closed. - // Soft failures (Tailscale CLI missing, peer not found, peer offline) fall - // back to the original URL so HTTP can report the connection error. - sinkURL := cfg.Sink.URL + + result["sinks"] = sinkReports + result["posted"] = len(results) > 0 && len(aggregateFailedLabels(results)) == 0 + _ = emit(result, strings.Join(humanLines, "\n")+"\n") + return results, dbsc, nil +} + +// aggregateFailedLabels returns the URLs of sinks whose push failed. +func aggregateFailedLabels(results []sinkResult) []string { + var failed []string + for _, r := range results { + if r.Err != nil { + failed = append(failed, r.URL) + } + } + return failed +} + +// resolveSinkSecret resolves the transport secret for a single sink. A sink +// with a peer requires that peer's key and never falls back to the legacy +// shared secret — a missing key isolates that sink rather than silently +// downgrading the payload to the fleet-wide shared secret. A peerless sink +// (the synthesized legacy single-sink case) uses the legacy shared secret. +func resolveSinkSecret(configDir string, sink config.SinkTarget, legacy string) (string, error) { + if sink.Peer != "" { + pk, err := keystore.Load(configDir, sink.Peer) + if err != nil { + return "", fmt.Errorf("no key for peer %q (run `agentcookie pair`): %w", sink.Peer, err) + } + return string(pk.Key), nil + } + if legacy == "" { + return "", fmt.Errorf("no transport credential: sink has no peer and no security.shared_secret") + } + return legacy, nil +} + +// postToSink resolves the sink URL through Tailscale (if needed) and POSTs +// the sealed payload, returning the sink's reply body. Bounded by the +// SyncClient timeout so a slow or dead sink cannot stall beyond its own +// window. +func postToSink(ctx context.Context, rawURL string, sealed []byte, verbose bool) (string, error) { + // Resolve sink URL hostname to IP via Tailscale if needed, so a + // MagicDNS hostname (http://grok-bot:9999/sync) works instead of a + // frozen 100.x IP. Fail-closed errors (ErrAmbiguousPeer) abort this + // sink; soft failures fall back to the raw URL so HTTP reports the + // connection error. + sinkURL := rawURL if resolved, resolveErr := resolveSinkURL(ctx, sinkURL); resolveErr != nil { if errors.Is(resolveErr, tsclient.ErrAmbiguousPeer) { - return 0, dbsc, fmt.Errorf("resolve sink URL: %w", resolveErr) + return "", fmt.Errorf("resolve sink URL: %w", resolveErr) } - // Soft failure: Tailscale not available, peer offline, etc. - // Fall back to the original URL and let HTTP report the error. if verbose { fmt.Fprintf(os.Stderr, "agentcookie source: sink URL resolution failed (%v); using original %s\n", resolveErr, sinkURL) } @@ -417,34 +563,23 @@ func pushOnce( sinkURL = resolved } - // Bound the POST by the SyncClient profile's timeout (5 minutes - // in v0.12) so a heavy LocalStorage / IndexedDB payload over a - // slow tailnet link does not get cut off at the pre-v0.12 30s - // floor. The Client.Timeout itself still applies; context.Done - // is just the cooperative path that gives the handler a clean - // cancellation. postCtx, cancel := context.WithTimeout(ctx, httpserver.Defaults(httpserver.SyncClient).ClientTimeout) defer cancel() req, err := http.NewRequestWithContext(postCtx, "POST", sinkURL, bytes.NewReader(sealed)) if err != nil { - return 0, dbsc, fmt.Errorf("new request: %w", err) + return "", fmt.Errorf("new request: %w", err) } req.Header.Set("Content-Type", "application/octet-stream") resp, err := httpserver.Client(httpserver.SyncClient).Do(req) if err != nil { - return 0, dbsc, fmt.Errorf("POST to sink %s: %w", sinkURL, err) + return "", fmt.Errorf("POST to sink %s: %w", sinkURL, err) } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) - result["posted"] = resp.StatusCode == http.StatusOK - result["sink_response"] = string(body) - result["sink_status"] = resp.StatusCode - if resp.StatusCode != http.StatusOK { - return 0, dbsc, fmt.Errorf("sink returned %d: %s", resp.StatusCode, string(body)) + return "", fmt.Errorf("sink returned %d: %s", resp.StatusCode, string(body)) } - _ = emit(result, fmt.Sprintf("agentcookie source: posted %d cookies, sink replied: %s%s\n", len(all), string(body), dbscNote(dbsc))) - return len(all), dbsc, nil + return string(body), nil } // dbscNote returns a concise " (N DBSC-suspect: warned/skipped)" suffix for the diff --git a/internal/cli/source_test.go b/internal/cli/source_test.go index 70d95fa..8b8be88 100644 --- a/internal/cli/source_test.go +++ b/internal/cli/source_test.go @@ -231,8 +231,9 @@ func newSourcePushFixture(t *testing.T, cookies []chrome.Cookie) *sourcePushFixt t.Cleanup(func() { http.DefaultTransport = oldTransport }) cfg := &config.SourceConfig{ - Sink: config.SinkRef{URL: "http://agentcookie-sink.test/sync"}, - Chrome: config.ChromeRef{DBPath: dbPath}, + Sink: config.SinkRef{URL: "http://agentcookie-sink.test/sync"}, + Chrome: config.ChromeRef{DBPath: dbPath}, + Security: config.SecurityRef{SharedSecret: secret}, } return &sourcePushFixture{ configDir: configDir, @@ -245,7 +246,7 @@ func newSourcePushFixture(t *testing.T, cookies []chrome.Cookie) *sourcePushFixt } func (f *sourcePushFixture) push() (int, error) { - return pushWithFreshBlocklist(context.Background(), f.cfg, f.key, f.secret, false, false, false, f.srcState, nil) + return pushWithFreshBlocklist(context.Background(), f.cfg, f.key, false, false, false, f.srcState, nil) } func (f *sourcePushFixture) batchCount() int { @@ -260,6 +261,10 @@ type sourceCapture struct { secret string mu sync.Mutex batches [][]chrome.Cookie + urls []string + // failURL, when non-empty, makes a POST to that exact URL return a + // non-200 so tests can exercise per-sink failure isolation. + failURL string } func newSourceCapture(t *testing.T, secret string) *sourceCapture { @@ -271,6 +276,15 @@ func (c *sourceCapture) RoundTrip(req *http.Request) (*http.Response, error) { if req.Method != http.MethodPost { return nil, fmt.Errorf("unexpected method %s", req.Method) } + if c.failURL != "" && req.URL.String() == c.failURL { + return &http.Response{ + StatusCode: http.StatusInternalServerError, + Status: "500 Internal Server Error", + Header: make(http.Header), + Body: io.NopCloser(bytes.NewBufferString("boom\n")), + Request: req, + }, nil + } sealed, err := io.ReadAll(req.Body) if err != nil { return nil, fmt.Errorf("read body: %w", err) @@ -285,6 +299,7 @@ func (c *sourceCapture) RoundTrip(req *http.Request) (*http.Response, error) { } c.mu.Lock() c.batches = append(c.batches, append([]chrome.Cookie(nil), envelope.Cookies...)) + c.urls = append(c.urls, req.URL.String()) c.mu.Unlock() return &http.Response{ @@ -296,6 +311,12 @@ func (c *sourceCapture) RoundTrip(req *http.Request) (*http.Response, error) { }, nil } +func (c *sourceCapture) postedURLs() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.urls...) +} + func (c *sourceCapture) batchCount() int { c.mu.Lock() defer c.mu.Unlock() @@ -426,3 +447,107 @@ func TestSourcePushSoftResolveErrorFallsBackToHostname(t *testing.T) { t.Fatalf("soft resolve error should still POST, got %d batches", got) } } + +// --- Multi-sink fan-out (U2) and per-sink state (U3) --- + +func TestSourcePushFansOutToMultipleSinks(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + // Two peerless sinks sharing the legacy secret (fixture bypasses + // LoadSource, which would otherwise require a per-sink credential). + urlA := "http://sink-a.test/sync" + urlB := "http://sink-b.test/sync" + fx.cfg.Sink = config.SinkRef{} + fx.cfg.Sinks = []config.SinkTarget{{URL: urlA}, {URL: urlB}} + + if _, err := fx.push(); err != nil { + t.Fatalf("push: %v", err) + } + if got := fx.batchCount(); got != 2 { + t.Fatalf("expected 2 POSTs (one per sink), got %d", got) + } + posted := fx.capture.postedURLs() + seen := map[string]bool{} + for _, u := range posted { + seen[u] = true + } + if !seen[urlA] || !seen[urlB] { + t.Fatalf("both sinks should have received the payload, got %v", posted) + } + // Per-sink state: two records, both with a successful push. + if n := len(fx.srcState.Sinks); n != 2 { + t.Fatalf("expected 2 per-sink state records, got %d", n) + } + for _, ps := range fx.srcState.Sinks { + if ps.TotalPushes != 1 || ps.TotalFailures != 0 { + t.Errorf("sink %s: expected 1 push 0 failures, got %+v", ps.URL, ps) + } + } +} + +func TestSourcePushIsolatesFailedSink(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 + + _, err := fx.push() + if err == nil { + t.Fatal("expected partial-failure error, got nil") + } + // The healthy sink still received the payload. + posted := fx.capture.postedURLs() + if len(posted) != 1 || posted[0] != urlGood { + t.Fatalf("healthy sink should have been delivered exactly once, got %v", posted) + } + // Per-sink state records the failure on the bad sink, success on the good. + var bad, good *state.SinkPushState + for i := range fx.srcState.Sinks { + switch fx.srcState.Sinks[i].URL { + case urlBad: + bad = &fx.srcState.Sinks[i] + case urlGood: + good = &fx.srcState.Sinks[i] + } + } + if bad == nil || bad.TotalFailures != 1 { + t.Errorf("bad sink should have 1 failure, got %+v", bad) + } + if good == nil || good.TotalPushes != 1 { + t.Errorf("good sink should have 1 push, got %+v", good) + } +} + +func TestSourcePushMissingKeyIsolatedNoSilentDowngrade(t *testing.T) { + fx := newSourcePushFixture(t, []chrome.Cookie{ + {HostKey: ".example.com", Name: "s", Value: "v", Path: "/"}, + }) + // One peer'd sink whose key is absent from the keystore, alongside a + // healthy peerless sink covered by the legacy shared secret. The peer'd + // sink must NOT be sealed under the shared secret (no silent downgrade): + // it fails, and only the peerless sink is POSTed. + urlPeer := "http://sink-peer.test/sync" + urlShared := "http://sink-shared.test/sync" + fx.cfg.Sink = config.SinkRef{} + fx.cfg.Sinks = []config.SinkTarget{ + {URL: urlPeer, Peer: "no-such-peer"}, + {URL: urlShared}, + } + + _, err := fx.push() + if err == nil { + t.Fatal("expected error for missing per-sink key, got nil") + } + if !strings.Contains(err.Error(), "no-such-peer") { + t.Errorf("error should name the peer with the missing key, got: %v", err) + } + posted := fx.capture.postedURLs() + if len(posted) != 1 || posted[0] != urlShared { + t.Fatalf("only the shared-secret sink should be posted (no downgrade of the peer'd sink), got %v", posted) + } +} diff --git a/internal/state/state.go b/internal/state/state.go index c4a260a..f047d17 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -31,6 +31,50 @@ type SourceState struct { LastDBSCWarned int `json:"last_dbsc_warned,omitempty"` LastDBSCSkipped int `json:"last_dbsc_skipped,omitempty"` LastDBSCSample []string `json:"last_dbsc_sample,omitempty"` + + // Sinks holds one record per fan-out sink (multi-sink). The top-level + // LastPush/TotalPushes/TotalFailures/LastError fields above remain the + // cross-sink aggregate so pre-multi-sink readers keep working; per-sink + // detail lives here. Empty on a single legacy sink or an old state file + // written before multi-sink, in which case the top-level fields and the + // legacy SinkURL are the whole story. + Sinks []SinkPushState `json:"sinks,omitempty"` +} + +// SinkPushState is one fan-out sink's push record, keyed by peer hostname +// (the stable pairing identity). URL is stored for display and updated in +// place when a sink's URL changes, so a URL edit does not orphan the record +// or lose its history. +type SinkPushState struct { + Peer string `json:"peer,omitempty"` + URL string `json:"url"` + LastPush time.Time `json:"last_push,omitempty"` + LastPushCount int `json:"last_push_count"` + TotalPushes int `json:"total_pushes"` + TotalFailures int `json:"total_failures"` + LastError string `json:"last_error,omitempty"` + LastErrorAt time.Time `json:"last_error_at,omitempty"` +} + +// SinkFor returns the per-sink record for a peer (or URL when the sink has +// no peer), creating and appending it on first use. Keying on peer keeps the +// record stable across a URL change; the record's URL is refreshed to the +// current value on each call so display stays accurate. +func (s *SourceState) SinkFor(peer, url string) *SinkPushState { + key := peer + for i := range s.Sinks { + match := s.Sinks[i].Peer == peer + if key == "" { + // Peerless sink: fall back to URL identity. + match = s.Sinks[i].Peer == "" && s.Sinks[i].URL == url + } + if match { + s.Sinks[i].URL = url + return &s.Sinks[i] + } + } + s.Sinks = append(s.Sinks, SinkPushState{Peer: peer, URL: url}) + return &s.Sinks[len(s.Sinks)-1] } // SinkState is the sink daemon's observable state, written on every accepted diff --git a/internal/state/state_test.go b/internal/state/state_test.go index 66d871e..f7dda7a 100644 --- a/internal/state/state_test.go +++ b/internal/state/state_test.go @@ -1,6 +1,7 @@ package state import ( + "os" "path/filepath" "sync" "testing" @@ -95,3 +96,44 @@ func TestWriterIsConcurrencySafe(t *testing.T) { } // Any save's value is acceptable; just confirm the file is valid JSON. } + +func TestSinkForKeysByPeerAndRefreshesURL(t *testing.T) { + s := &SourceState{Role: "source"} + a := s.SinkFor("alpha", "http://a.test/sync") + a.TotalPushes = 3 + // Same peer, changed URL: same record, URL refreshed, history kept. + again := s.SinkFor("alpha", "http://a-new.test/sync") + if again.TotalPushes != 3 { + t.Fatalf("expected history preserved across URL change, got %+v", again) + } + if again.URL != "http://a-new.test/sync" { + t.Errorf("URL should refresh to current, got %q", again.URL) + } + if len(s.Sinks) != 1 { + t.Fatalf("URL change must not orphan the record, got %d records", len(s.Sinks)) + } + // A different peer is a distinct record. + s.SinkFor("bravo", "http://b.test/sync") + if len(s.Sinks) != 2 { + t.Fatalf("distinct peer should add a record, got %d", len(s.Sinks)) + } +} + +func TestLoadSourceLegacySinkURLDecodesWithoutError(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "source-state.json") + // A pre-multi-sink state file: sink_url present, no sinks array. + if err := os.WriteFile(path, []byte(`{"role":"source","sink_url":"http://legacy.test/sync","total_pushes":5}`), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + st, err := LoadSource(path) + if err != nil { + t.Fatalf("legacy state should decode without error: %v", err) + } + if st.SinkURL != "http://legacy.test/sync" || st.TotalPushes != 5 { + t.Errorf("legacy fields lost: %+v", st) + } + if len(st.Sinks) != 0 { + t.Errorf("legacy file has no per-sink records, got %d", len(st.Sinks)) + } +} From 74891b3ffda2ebeb76600e142da1d8a29d951998 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:26:00 -0700 Subject: [PATCH 3/7] feat(cli): report per-sink push state in status and doctor status prints a per-sink breakdown (pushes, failures, last push, last error) under the aggregate source-daemon line when multi-sink fan-out records are present. doctor's source-state check names the failing sink(s) on a WARN and notes the healthy sink count, while the aggregate OK/WARN thresholds are unchanged. Single-sink and legacy state files render exactly as before. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- internal/cli/doctor.go | 28 ++++++++++++++++++++++++++-- internal/cli/doctor_test.go | 33 +++++++++++++++++++++++++++++++++ internal/cli/status.go | 18 ++++++++++++++++++ 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/internal/cli/doctor.go b/internal/cli/doctor.go index 7384f95..9d087d4 100644 --- a/internal/cli/doctor.go +++ b/internal/cli/doctor.go @@ -985,18 +985,42 @@ func checkSourceStateFrom(st *state.SourceState, err error) Check { } } if st.TotalFailures > 0 { + detail := fmt.Sprintf("last push %s ago, %d total failures", age, st.TotalFailures) + if failing := failingSinkLabels(st); failing != "" { + detail += fmt.Sprintf(" (failing sink(s): %s)", failing) + } return Check{ Name: "Source state", Severity: SeverityWarn, - Detail: fmt.Sprintf("last push %s ago, %d total failures", age, st.TotalFailures), + Detail: detail, Remediation: "inspect `agentcookie status` for the most recent error", } } + detail := fmt.Sprintf("last push %s ago, 0 failures", age) + if n := len(st.Sinks); n > 1 { + detail += fmt.Sprintf(", %d sinks all healthy", n) + } return Check{ Name: "Source state", Severity: SeverityOK, - Detail: fmt.Sprintf("last push %s ago, 0 failures", age), + Detail: detail, + } +} + +// failingSinkLabels lists the peer/URL of each fan-out sink whose most +// recent push failed, for the per-sink detail on the source-state WARN. +func failingSinkLabels(st *state.SourceState) string { + var labels []string + for _, s := range st.Sinks { + if s.TotalFailures > 0 && s.LastError != "" { + label := s.URL + if s.Peer != "" { + label = s.Peer + } + labels = append(labels, label) + } } + return strings.Join(labels, ", ") } // checkDBSCFrom is informational: it surfaces how many cookies the last push diff --git a/internal/cli/doctor_test.go b/internal/cli/doctor_test.go index 7c46aeb..b8e91cf 100644 --- a/internal/cli/doctor_test.go +++ b/internal/cli/doctor_test.go @@ -324,6 +324,39 @@ func TestCheckSourceState(t *testing.T) { t.Fatalf("got %q", c.Severity) } }) + t.Run("per-sink failure names the failing sink", func(t *testing.T) { + st := &state.SourceState{ + LastPush: time.Now(), + TotalFailures: 1, + Sinks: []state.SinkPushState{ + {Peer: "alpha", URL: "http://a.test/sync", TotalPushes: 4}, + {Peer: "bravo", URL: "http://b.test/sync", TotalFailures: 1, LastError: "connection refused"}, + }, + } + c := checkSourceStateFrom(st, nil) + if c.Severity != SeverityWarn { + t.Fatalf("got %q", c.Severity) + } + if !strings.Contains(c.Detail, "bravo") { + t.Errorf("detail should name the failing sink, got %q", c.Detail) + } + }) + t.Run("healthy multi-sink notes sink count", func(t *testing.T) { + st := &state.SourceState{ + LastPush: time.Now(), + Sinks: []state.SinkPushState{ + {Peer: "alpha", URL: "http://a.test/sync", TotalPushes: 4}, + {Peer: "bravo", URL: "http://b.test/sync", TotalPushes: 4}, + }, + } + c := checkSourceStateFrom(st, nil) + if c.Severity != SeverityOK { + t.Fatalf("got %q (%q)", c.Severity, c.Detail) + } + if !strings.Contains(c.Detail, "2 sinks") { + t.Errorf("detail should note the sink count, got %q", c.Detail) + } + }) } func TestCheckDBSC(t *testing.T) { diff --git a/internal/cli/status.go b/internal/cli/status.go index 49ea96e..89b7623 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -138,6 +138,24 @@ var statusCmd = &cobra.Command{ } fmt.Printf(" source daemon: %d pushes, %d failures, last push %s\n", st.SourceState.TotalPushes, st.SourceState.TotalFailures, ago) + // Per-sink breakdown (multi-sink fan-out). Absent for a + // single legacy sink or an old state file, where the + // aggregate line above is the whole story. + for _, sink := range st.SourceState.Sinks { + label := sink.URL + if sink.Peer != "" { + label = fmt.Sprintf("%s (%s)", sink.Peer, sink.URL) + } + sinkAgo := "never" + if !sink.LastPush.IsZero() { + sinkAgo = time.Since(sink.LastPush).Round(time.Second).String() + " ago" + } + fmt.Printf(" sink %s: %d pushes, %d failures, last push %s\n", + label, sink.TotalPushes, sink.TotalFailures, sinkAgo) + if sink.LastError != "" { + fmt.Printf(" last error: %s\n", sink.LastError) + } + } } if st.SinkState != nil { ago := "never" From 5d202b0a2203d20180079c7b63e355261ec8d67d Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:29:23 -0700 Subject: [PATCH 4/7] feat(wizard): add --add-sink to pair and append an extra fan-out sink --add-sink pairs an additional peer and appends it to an existing source.yaml as a multi-sink target, migrating a legacy single-sink config into an explicit sinks: list, instead of overwriting. It validates against a duplicate peer or URL before pairing, files the new key under the operator --peer name via beginSourcePairing (the announced hostname is recorded separately), and rewrites source.yaml via a template rather than yaml.Marshal so no stray empty legacy sink: block is emitted. The running --watch daemon fans out to the new sink on its next push, so no LaunchAgent reinstall is needed. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- internal/cli/wizard.go | 136 ++++++++++++++++++++++++++++++++++++ internal/cli/wizard_test.go | 64 +++++++++++++++++ 2 files changed, 200 insertions(+) diff --git a/internal/cli/wizard.go b/internal/cli/wizard.go index c011980..6e834f4 100644 --- a/internal/cli/wizard.go +++ b/internal/cli/wizard.go @@ -46,6 +46,7 @@ var ( wizardWriteChromeSQLite bool wizardNoCDP bool wizardNoCmux bool + wizardAddSink bool ) var wizardCmd = &cobra.Command{ @@ -108,6 +109,7 @@ func init() { wizardInstallCmd.Flags().BoolVar(&wizardWriteChromeSQLite, "write-chrome-sqlite", false, "[sink] force universal delivery (write the real Default Chrome profile) and honor it even if the one-password keychain open cannot complete; does not silently downgrade to degraded") wizardInstallCmd.Flags().BoolVar(&wizardNoCDP, "no-cdp", false, "[sink] do not enable CDP injection alongside skip_chrome_sqlite. By default, headless installs enable CDP injection so Chrome on the sink still sees synced cookies. Pass --no-cdp for sidecar+adapter-only mode.") wizardInstallCmd.Flags().BoolVar(&wizardNoCmux, "no-cmux", false, "do not auto-enable the cmux local loop even if cmux is installed (by default, install wires Chrome->cmux delivery when cmux is present)") + wizardInstallCmd.Flags().BoolVar(&wizardAddSink, "add-sink", false, "[source] pair an ADDITIONAL sink and append it to an existing source.yaml (multi-sink fan-out) instead of overwriting; requires an existing source config") wizardUninstallCmd.Flags().StringVar(&wizardRole, "as", "", "source | sink (required)") wizardUninstallCmd.Flags().BoolVar(&wizardForce, "purge", false, "also delete configs and paired keys") @@ -157,6 +159,14 @@ func wizardInstallSource(ctx context.Context, binPath, logDir string) error { return err } + // --add-sink: pair an ADDITIONAL peer and append it to an existing + // source.yaml as a multi-sink fan-out target, instead of the default + // overwrite-or-create flow. The running --watch daemon picks up the + // new sink on its next push, so no LaunchAgent reinstall is needed. + if wizardAddSink { + return wizardAddSinkToSource(ctx, binPath, logDir) + } + // Step 1: drop source.yaml + blocklist.yaml if missing or force. // v0.12.0-beta.2: if source.yaml already exists with a peer.hostname // that differs from --peer, fail loud rather than silently keeping the @@ -507,6 +517,67 @@ func runWizardUninstall(cmd *cobra.Command, args []string) error { return nil } +// wizardAddSinkToSource pairs an additional peer and appends it to an +// existing source.yaml as a multi-sink fan-out target. It requires an +// existing source config, validates the new sink is not a duplicate before +// pairing, files the new key under the operator --peer name (via +// beginSourcePairing), then rewrites source.yaml with the appended sinks +// list. The running --watch daemon fans out to it on the next push. +func wizardAddSinkToSource(ctx context.Context, binPath, logDir string) error { + sourcePath := filepath.Join(common.ConfigDir, "source.yaml") + if !fileExists(sourcePath) { + return fmt.Errorf("--add-sink needs an existing source.yaml; run `agentcookie wizard install --as source` first") + } + if wizardPeer == "" { + return fmt.Errorf("--add-sink requires --peer (the new sink's hostname)") + } + cfg, err := config.LoadSource(common.ConfigDir) + if err != nil { + return fmt.Errorf("load existing source.yaml: %w", err) + } + newURL := wizardSinkURL + if newURL == "" { + newURL = fmt.Sprintf("http://%s:9999/sync", wizardPeer) + } + // Validate (reject duplicate peer/URL) BEFORE pairing so a bad add + // does not leave an orphaned key behind. + newYAML, err := buildAddSinkYAML(cfg, newURL, wizardPeer) + if err != nil { + return err + } + + // Pair the new peer. beginSourcePairing files the key under the + // operator --peer name (with the announced hostname recorded + // separately), which is the name buildAddSinkYAML wrote into sinks. + keyPath, _ := keystore.Path(common.ConfigDir, wizardPeer) + if fileExists(keyPath) && !wizardRepair { + fmt.Fprintf(os.Stderr, "agentcookie wizard: existing paired key for %q found; skipping pairing (use --repair to force)\n", wizardPeer) + } else { + listen := wizardListen + if listen == "" { + ip, err := tsclient.RequireTailnetIP(ctx) + if err != nil { + return fmt.Errorf("detect Tailscale 100.x address for pair listener: %w", err) + } + listen = fmt.Sprintf("%s:9998", ip) + } else if err := validateListenAddr(listen); err != nil { + return fmt.Errorf("--listen %q: %w", listen, err) + } + pairingInfo, code, err := beginSourcePairing(ctx, listen, wizardLocalName, binPath, logDir) + if err != nil { + return fmt.Errorf("pairing: %w", err) + } + fmt.Fprintln(os.Stderr, pairingInfo) + fmt.Fprintf(os.Stderr, "agentcookie wizard: paired additional sink %q (code was %s)\n", wizardPeer, code) + } + + if err := os.WriteFile(sourcePath, []byte(newYAML), 0o600); err != nil { + return fmt.Errorf("write updated source.yaml: %w", err) + } + fmt.Fprintf(os.Stderr, "agentcookie wizard: appended sink %s (peer %q) to source.yaml; the --watch daemon will fan out to it on the next push\n", newURL, wizardPeer) + return nil +} + // beginSourcePairing starts a source-side pairing listener and waits for the // sink to connect. Returns a human-readable instruction block (which is also // the content of ~/.agentcookie/pairing.json) plus the code, blocking until @@ -681,6 +752,71 @@ peer: `, sinkURL, peer) } +// renderSourceYAMLSinks renders a multi-sink source.yaml body, always as an +// explicit sinks: list (never the legacy scalar sink:/peer:), preserving the +// loaded config's chrome / browser / security / cmux settings. Written via +// template rather than yaml.Marshal on purpose: yaml.v3 does not omit a +// zero-value legacy Sink struct even with omitempty, so marshaling a +// sinks-only config would emit a stray empty sink: block. +func renderSourceYAMLSinks(cfg *config.SourceConfig, sinks []config.SinkTarget) string { + var b strings.Builder + b.WriteString("sinks:\n") + for _, s := range sinks { + b.WriteString(fmt.Sprintf(" - url: %s\n", s.URL)) + if s.Peer != "" { + b.WriteString(fmt.Sprintf(" peer: %s\n", s.Peer)) + } + } + dbPath := cfg.Chrome.DBPath + if dbPath == "" { + dbPath = "~/Library/Application Support/Google/Chrome/Default/Cookies" + } + b.WriteString("chrome:\n") + b.WriteString(fmt.Sprintf(" db_path: %s\n", dbPath)) + if cfg.Browser.Name != "" || cfg.Browser.Profile != "" { + b.WriteString("browser:\n") + if cfg.Browser.Name != "" { + b.WriteString(fmt.Sprintf(" name: %s\n", cfg.Browser.Name)) + } + if cfg.Browser.Profile != "" { + b.WriteString(fmt.Sprintf(" profile: %s\n", cfg.Browser.Profile)) + } + } + if cfg.Security.SharedSecret != "" { + b.WriteString("security:\n") + b.WriteString(fmt.Sprintf(" shared_secret: %s\n", cfg.Security.SharedSecret)) + } + if cfg.Cmux.Enabled { + b.WriteString("cmux:\n") + b.WriteString(fmt.Sprintf(" enabled: %v\n", cfg.Cmux.Enabled)) + if cfg.Cmux.CmuxPath != "" { + b.WriteString(fmt.Sprintf(" cmux_path: %s\n", cfg.Cmux.CmuxPath)) + } + } + return b.String() +} + +// buildAddSinkYAML computes the new source.yaml body when adding a sink to an +// already-loaded source config. It migrates a legacy single-sink config into +// an explicit sinks: list and appends the new target, rejecting a duplicate +// peer or URL so --add-sink is idempotent-safe. +func buildAddSinkYAML(cfg *config.SourceConfig, newURL, newPeer string) (string, error) { + if newURL == "" { + return "", fmt.Errorf("a sink URL is required (set --sink-url or it defaults from --peer)") + } + sinks := append([]config.SinkTarget(nil), cfg.ResolvedSinks()...) + for _, s := range sinks { + if newPeer != "" && s.Peer == newPeer { + return "", fmt.Errorf("a sink for peer %q is already configured", newPeer) + } + if s.URL == newURL { + return "", fmt.Errorf("a sink with URL %q is already configured", newURL) + } + } + sinks = append(sinks, config.SinkTarget{URL: newURL, Peer: newPeer}) + return renderSourceYAMLSinks(cfg, sinks), nil +} + // renderSinkYAML formats sink.yaml with a caller-resolved listen // address. The wizard install path is the only place that calls this, // and it resolves listenAddr via tsclient.RequireTailnetIP first so a diff --git a/internal/cli/wizard_test.go b/internal/cli/wizard_test.go index cc9a159..8640715 100644 --- a/internal/cli/wizard_test.go +++ b/internal/cli/wizard_test.go @@ -7,6 +7,8 @@ import ( "path/filepath" "strings" "testing" + + "github.com/mvanhorn/agentcookie/internal/config" ) // errInjectedKeychainOpen is the failure injected into the @@ -535,3 +537,65 @@ func TestSkipChromeSQLiteHelpNotStale(t *testing.T) { t.Errorf("help should describe the degraded opt-out: %q", f.Usage) } } + +func TestBuildAddSinkYAMLMigratesLegacyAndAppends(t *testing.T) { + cfg := &config.SourceConfig{ + Sink: config.SinkRef{URL: "http://first.test:9999/sync"}, + Peer: config.PeerRef{Hostname: "first"}, + Chrome: config.ChromeRef{DBPath: "/tmp/Cookies"}, + } + yamlBody, err := buildAddSinkYAML(cfg, "http://second.test:9999/sync", "second") + if err != nil { + t.Fatalf("buildAddSinkYAML: %v", err) + } + // The rendered YAML must decode through the real loader into two sinks. + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "source.yaml"), []byte(yamlBody), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + loaded, err := config.LoadSource(dir) + if err != nil { + t.Fatalf("LoadSource on rendered YAML: %v", err) + } + got := loaded.ResolvedSinks() + if len(got) != 2 { + t.Fatalf("expected 2 sinks after add, got %d (%s)", len(got), yamlBody) + } + if got[0].Peer != "first" || got[1].Peer != "second" { + t.Errorf("sinks wrong: %+v", got) + } + if strings.Contains(yamlBody, "\nsink:\n") { + t.Errorf("rendered YAML must not emit a legacy sink: block:\n%s", yamlBody) + } +} + +func TestBuildAddSinkYAMLRejectsDuplicatePeer(t *testing.T) { + cfg := &config.SourceConfig{ + Sinks: []config.SinkTarget{{URL: "http://a.test/sync", Peer: "alpha"}}, + } + if _, err := buildAddSinkYAML(cfg, "http://a2.test/sync", "alpha"); err == nil { + t.Fatal("expected error for duplicate peer, got nil") + } + if _, err := buildAddSinkYAML(cfg, "http://a.test/sync", "beta"); err == nil { + t.Fatal("expected error for duplicate URL, got nil") + } +} + +func TestRenderSourceYAMLSinksPreservesSharedSecret(t *testing.T) { + cfg := &config.SourceConfig{ + Chrome: config.ChromeRef{DBPath: "/tmp/Cookies"}, + Security: config.SecurityRef{SharedSecret: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}, + } + body := renderSourceYAMLSinks(cfg, []config.SinkTarget{{URL: "http://a.test/sync"}}) + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "source.yaml"), []byte(body), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + loaded, err := config.LoadSource(dir) + if err != nil { + t.Fatalf("LoadSource: %v (body:\n%s)", err, body) + } + if loaded.Security.SharedSecret == "" { + t.Error("shared secret not preserved through render") + } +} From 4811c0c128f5c28141eca2c0e69b474126d7c780 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:31:05 -0700 Subject: [PATCH 5/7] docs: document multi-sink fan-out Add a README section on fanning out to multiple sinks (sinks: list, --add-sink, per-sink status/doctor, the equal-trust caveat and manual revoke), a CHANGELOG entry, a dedicated examples/source-multi-sink.yaml, and a pointer from examples/source.yaml. Guard the example against schema drift with a loader decode test. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- CHANGELOG.md | 13 +++++++++++++ README.md | 28 ++++++++++++++++++++++++++++ examples/source-multi-sink.yaml | 32 ++++++++++++++++++++++++++++++++ examples/source.yaml | 4 ++++ internal/config/config_test.go | 20 ++++++++++++++++++++ 5 files changed, 97 insertions(+) create mode 100644 examples/source-multi-sink.yaml diff --git a/CHANGELOG.md b/CHANGELOG.md index a5031cc..bba3b66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## [Unreleased] + +### Multi-sink fan-out + +One source can now push the same cookies and secrets to several sinks. + +- `source.yaml` accepts a `sinks:` list, each entry with its own `url` and `peer`. A legacy single-sink config (`sink:` + `peer:`) keeps working unchanged and loads as a one-element list. +- A push reads and filters cookies once, then seals and POSTs to each sink with its own paired key. A missing per-sink key isolates that sink instead of aborting the whole push, and never silently downgrades a paired sink to the legacy shared secret. +- Per-sink failures are isolated; a partial or total sink failure is a non-zero `--once` exit. The `--once` deadline scales with sink count so a slow first sink cannot starve later healthy ones. +- `agentcookie wizard install --as source --add-sink --peer --sink-url ` pairs and appends an additional sink to an existing config. +- `agentcookie status` and `agentcookie doctor` report per-sink push state. +- Example: `examples/source-multi-sink.yaml`. + ## [1.0.0] - 2026-08-13 ### Featured: Mac to Linux continuous sync diff --git a/README.md b/README.md index 396dfad..99ccbc2 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,34 @@ agentcookie wizard install --as sink \ The macOS sink writes to Chrome's encrypted SQLite, the plaintext sidecar, and per-CLI adapter session files. It can also run CDP injection into a managed Chrome subprocess. See [docs/quickstart.md](docs/quickstart.md) for the full macOS-to-macOS walkthrough. +## Fan out to multiple sinks + +One source can push the same cookies and secrets to several sinks. A source push reads and filters your cookies once, then seals and POSTs to each sink independently with that sink's own paired key. A sink that is down or unreachable is isolated: it fails on its own while the other sinks still receive the payload. + +List sinks under `sinks:` in `source.yaml`, each with its own `url` and `peer`: + +```yaml +sinks: + - url: http://mac-mini.tailnet.ts.net:9999/sync + peer: mac-mini + - url: http://grok-bot.tailnet.ts.net:9999/sync + peer: grok-bot +chrome: + db_path: ~/Library/Application Support/Google/Chrome/Default/Cookies +``` + +A legacy single-sink `source.yaml` (top-level `sink:` + `peer:`) keeps working unchanged and behaves as a one-element list. To pair and append another sink without hand-editing: + +```bash +agentcookie wizard install --as source --add-sink \ + --peer \ + --sink-url http://.tailnet.ts.net:9999/sync +``` + +`agentcookie status` and `agentcookie doctor` report each sink's last push and failures. See [examples/source-multi-sink.yaml](examples/source-multi-sink.yaml). + +**Trust note:** every sink receives the same full cookie and secret set, so a compromise of the least-trusted sink exposes everything. Only list sinks you trust with the whole payload. To stop feeding a sink, remove its entry from `sinks:` and delete its `keys/.json`. The device-bound (DBSC) caveat below is per-sink and unchanged: each sink still needs its own Chrome signed into the same Google account. + ## What about Chrome's device-bound cookies (DBSC)? Chrome's Device Bound Session Credentials (DBSC) tie a session to one machine's secure hardware so a stolen cookie cannot be replayed elsewhere. For a site that has adopted DBSC, a copied cookie works on the sink only until its short-lived window (minutes) lapses. diff --git a/examples/source-multi-sink.yaml b/examples/source-multi-sink.yaml new file mode 100644 index 0000000..e4df24b --- /dev/null +++ b/examples/source-multi-sink.yaml @@ -0,0 +1,32 @@ +# Example multi-sink source.yaml. Copy to ~/.config/agentcookie/source.yaml +# on the machine where you log in interactively (your laptop) to fan out the +# same cookies and secrets to more than one sink. +# +# One source push reads and filters your cookies once, then seals and POSTs +# to each sink independently. A sink that is down or unreachable is isolated: +# it fails on its own while the other sinks still receive the payload. +# +# Trust note: every sink receives the SAME full cookie and secret set, so a +# compromise of the least-trusted sink exposes everything. Only list sinks you +# trust with the whole payload. + +sinks: + # Each entry has its own URL and peer (the keys/.json filename from + # `agentcookie pair`). Prefer MagicDNS hostnames over 100.x IPs so the + # source follows each sink across Tailscale re-auth. + - url: http://mac-mini.tailnet.ts.net:9999/sync + peer: mac-mini + - url: http://grok-bot.tailnet.ts.net:9999/sync + peer: grok-bot + +chrome: + # Defaults to ~/Library/Application Support/Google/Chrome/Default/Cookies + # if omitted. Set explicitly to read from a non-default profile. + db_path: ~/Library/Application Support/Google/Chrome/Default/Cookies + +# A legacy single-sink source.yaml (top-level `sink:` + `peer:`) still works +# unchanged and behaves as a one-element sinks list. Pair and append another +# sink without hand-editing with: +# +# agentcookie wizard install --as source --add-sink --peer \ +# --sink-url http://.tailnet.ts.net:9999/sync diff --git a/examples/source.yaml b/examples/source.yaml index 474cbc3..4339845 100644 --- a/examples/source.yaml +++ b/examples/source.yaml @@ -31,3 +31,7 @@ peer: # who haven't paired yet. After pairing, delete the field below entirely. # security: # shared_secret: only-set-this-before-you-pair + +# Multi-sink: to push the same cookies to more than one sink, use a +# top-level `sinks:` list instead of the single `sink:`/`peer:` above. +# See examples/source-multi-sink.yaml. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 7de6e13..40c8a83 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -824,3 +824,23 @@ chrome: t.Fatalf("shared secret should cover a peerless sink: %v", err) } } + +func TestExampleMultiSinkConfigDecodes(t *testing.T) { + // The shipped multi-sink example must load through the strict loader, + // so the docs never drift from the accepted schema. + dir := t.TempDir() + data, err := os.ReadFile(filepath.Join("..", "..", "examples", "source-multi-sink.yaml")) + if err != nil { + t.Fatalf("read example: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "source.yaml"), data, 0o600); err != nil { + t.Fatalf("write: %v", err) + } + cfg, err := LoadSource(dir) + if err != nil { + t.Fatalf("example source-multi-sink.yaml should load: %v", err) + } + if got := len(cfg.ResolvedSinks()); got != 2 { + t.Fatalf("example should declare 2 sinks, got %d", got) + } +} From d3b20b2afe362fced709f8abdbfc57e8feef24f5 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:31:33 -0700 Subject: [PATCH 6/7] docs(plan): add multi-sink cookie fan-out implementation plan Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- ...1436-feat-multi-sink-cookie-fanout-plan.md | 293 ++++++++++++++++++ 1 file changed, 293 insertions(+) create mode 100644 docs/plans/2026-09-09-1436-feat-multi-sink-cookie-fanout-plan.md diff --git a/docs/plans/2026-09-09-1436-feat-multi-sink-cookie-fanout-plan.md b/docs/plans/2026-09-09-1436-feat-multi-sink-cookie-fanout-plan.md new file mode 100644 index 0000000..65ad4d7 --- /dev/null +++ b/docs/plans/2026-09-09-1436-feat-multi-sink-cookie-fanout-plan.md @@ -0,0 +1,293 @@ +--- +artifact_contract: ce-unified-plan/v1 +artifact_readiness: implementation-ready +execution: code +product_contract_source: ce-plan-bootstrap +type: feat +created: 2026-09-09 +--- + +# feat: Fan out cookie and secret sync to multiple sinks + +## Summary + +`agentcookie source` pushes to exactly one sink. `SourceConfig` holds a single `Sink SinkRef` and a single `Peer PeerRef`, and `pushOnce` seals the payload with that one peer's key and POSTs once to `cfg.Sink.URL`. Matt now runs three sink machines (instinct/e2b, a Mac mini, and grok-bot) and can only feed one at a time by editing the URL and re-pushing, which the `--watch` daemon cannot do for more than one target. + +This plan adds native multi-sink fan-out. One source config gains a `sinks` list; the read-filter-seal pipeline reads and filters cookies once, then seals per sink with that sink's own peer key and POSTs to each, isolating a failed or unreachable sink so it never blocks the others. The existing single `sink`/`peer` fields stay valid and load as a one-element list, so every deployed config keeps working with no migration. Secrets ride the same sealed payload, so secret fan-out comes for free once the POST loops. + +--- + +## Problem Frame + +The push path is single-target by data model, not by accident: + +- `internal/config/config.go` — `SourceConfig.Sink` is one `SinkRef{URL}` and `SourceConfig.Peer` is one `PeerRef{Hostname}`. The hostname names a single key under `keys/`. +- `internal/cli/source.go` — `runSource` resolves one transport secret via `resolveTransportSecret(common.ConfigDir, cfg.Peer.Hostname, cfg.Security.SharedSecret)` (third arg is the legacy shared secret string, not the config). `pushOnce` builds the plaintext payload (cookies plus the secrets-bus payload) once, calls `transport.SealWithSecret(payload, secret)`, and POSTs once to `cfg.Sink.URL`. +- `internal/cli/common.go` — `resolveTransportSecret(configDir, peerHost, legacy string)` returns the peer key when present, else falls back to the legacy `Security.SharedSecret`, else errors. This fallback is load-bearing for multi-sink (see KTD2): a missing per-sink key must not silently seal under the fleet-wide shared secret. +- `internal/cli/wizard.go` — pairing files the key under the operator-supplied `--peer` name via `beginSourcePairing` / the pairing-info writer, recording the sink's announced hostname separately. `runPairAsSource` in `pair.go` instead files under `res.RemotePeer` (the announced `os.Hostname`), which is the wrong name for the fan-out lookup (this is the live "announced as e2b.local, stored as instinct" behavior). +- `internal/state/state.go` — `SourceState.SinkURL` is a single string, and the push counters (`LastPush`, `TotalPushes`, `TotalFailures`, `LastError`) are tracked for that one target. + +To send to three machines today, the only path is three separate config directories each with its own `source.yaml`, key, and `--watch` LaunchAgent. That works but triples the Chrome reads, seals, and daemons and splits status across three `doctor` runs. Matt confirmed he wants native multi-sink in the repo he owns, with existing single-sink configs left working. + +The device-bound (DBSC) cookie caveat is unchanged and per-sink by nature: the ~86 Google cookies pinned to the source Mac will not work on any sink. More sinks does not change that; each sink still needs its own Chrome signed into the same Google account. + +--- + +## Goal Capsule + +**Objective:** one `agentcookie source` config keeps N sink machines logged in, so Matt's fleet of sinks all receive cookies and secrets from one watcher without per-target config directories. + +**Means:** a `sinks` list in `SourceConfig`, a fan-out push loop that reads once and seals-and-POSTs per sink with per-sink failure isolation, additive wizard pairing to grow the list, and per-sink reporting in `status` and `doctor`. + +**Done when:** a config listing three sinks delivers to all three on one push, one dead sink does not fail the others, a legacy single-sink config still works unchanged, and `status`/`doctor` report each sink's last-push and reachability. + +--- + +## Requirements + +- **R1** — `SourceConfig` accepts a list of sinks, each carrying its own sink URL and peer hostname (its key). +- **R2** — A legacy `source.yaml` with the single `sink:`/`peer:` fields loads unchanged and behaves exactly as before (one-element list synthesized at load). +- **R3** — One push reads and filters cookies once, then seals per sink with that sink's peer key and POSTs to each sink's URL. Secrets ride the same payload, so each sink receives them too. +- **R4** — A sink that fails to seal or POST (unreachable, auth failure, timeout) is recorded as failed for that sink and does not abort delivery to the other sinks. The push exit status reflects whole-vs-partial success. +- **R5** — The wizard can add a sink to an existing source config additively (pair a new peer and append it) without overwriting the existing sinks. +- **R6** — `status` and `doctor` report per-sink state: URL, last push, last error, and reachability, instead of a single sink. +- **R7** — `--watch` fans out to all configured sinks on each Chrome cookie or secrets write. + +--- + +## Key Technical Decisions + +**KTD1. Additive config schema, legacy fields retained** *(session-settled: user-directed — chosen over a hard migration: keep every deployed source.yaml working with no rewrite).* Add `Sinks []SinkTarget` to `SourceConfig`, where `SinkTarget` carries `URL` plus its own `Peer` hostname. Keep the existing `Sink SinkRef` and `Peer PeerRef` fields. At load, when `Sinks` is empty and legacy `Sink.URL` is set, synthesize a one-element `Sinks` list from `Sink`+`Peer`. When `Sinks` is populated, it is authoritative and the legacy scalars are ignored. Marshaling writes `sinks` for multi-sink configs and may leave the legacy fields untouched on configs that still use them. Governs R1, R2. + +**KTD2. Read once, seal-and-POST per sink; resolve secret inside the loop** The plaintext payload (cookies after blocklist and DBSC filtering, plus the secrets-bus payload) is built once. The fan-out is only over sealing and transport: for each sink, resolve its secret with `resolveTransportSecret(common.ConfigDir, sink.Peer, cfg.Security.SharedSecret)`, `SealWithSecret` with it, and POST to its URL. Sealing must be per sink because each sink has a distinct pairing-derived key. **Resolve inside the fan-out loop, not up front:** `resolveTransportSecret` hard-errors on a missing key, so resolving all secrets before the loop would let one missing key abort delivery to every sink, defeating R4. A resolution failure is recorded as that sink's failure and the loop continues. **No silent shared-secret downgrade:** a per-sink key that is missing or unreadable must isolate that sink (recorded failure, skipped POST), never fall through to the fleet-wide `Security.SharedSecret` and seal the full payload under it. The shared-secret path stays available only for a legacy single-sink config whose synthesized sink has an empty `Peer` (see KTD1). Governs R3, R4. + +**KTD3. Sequential fan-out with a per-sink timeout, failures isolated, `--once` budget scaled** Iterate sinks in order; bound each POST by the existing per-sink `SyncClient` timeout (5 min) so one slow or dead sink cannot stall the rest beyond its own timeout. Collect a per-sink result (ok / error) and continue on error. The push returns success only if every sink succeeded; partial success is reported per sink and reflected in the exit status. + +**The `--once` outer deadline must scale with sink count.** `runSource` today wraps the whole push in one `context.WithTimeout(ctx, SyncClient.ClientTimeout + 30s)` (5m30s). Sequential fan-out shares that single budget, so a first sink that runs to its full 5-minute timeout starves sinks 2 and 3 and they fail even when healthy, defeating R4 in `--once` mode. Fix: in `--once`, either scale the outer context to roughly `len(sinks) * ClientTimeout + slack`, or drop the outer wrapper and rely solely on each POST's own `postCtx` bound. `--watch` has no outer per-push timeout, so this affects `--once` only. Governs R4. + +Concurrency across sinks is deferred (see Deferred to Follow-Up Work), but see Open Questions: a chronically-dead sink under `--watch` pays its full 5-minute timeout serially on every push, degrading freshness to the healthy sinks, so a fast-fail connect timeout or per-sink backoff may be needed before this is truly deferrable. Governs R4, R7. + +**KTD4. Per-sink source state, keyed by peer hostname** Replace the single `SinkURL` and shared counters in `SourceState` with per-sink records, each carrying last push time, last push count, totals, and last error. **Key by peer hostname, not URL:** the peer hostname is the stable pairing identity, whereas a sink URL is mutable (a tailnet IP or port edit would orphan a URL-keyed record and lose its history). Store the current URL as a field on the record so `status`/`doctor` can show it. Preserve the DBSC summary fields at the top level since the filtered cookie set is identical across sinks. `status` and `doctor` read the per-sink records. Governs R6. + +--- + +## Implementation Units + +### U1. Sinks list in SourceConfig with legacy synthesis + +**Goal:** the config layer represents N sinks and still loads every legacy single-sink config unchanged. + +**Requirements:** R1, R2. + +**Dependencies:** none. + +**Files:** +- `internal/config/config.go` — add `SinkTarget{URL, Peer}` and `Sinks []SinkTarget` on `SourceConfig`; keep legacy `Sink`/`Peer`. +- `internal/config/config_test.go` — load and marshal coverage. +- `examples/source.yaml` (and any multi-sink example added) — document the `sinks:` shape. + +**Approach:** +1. Add a `SinkTarget` type with `URL` (`yaml:"url"`) and `Peer` (`yaml:"peer"`, the key hostname). +2. Add `Sinks []SinkTarget` (`yaml:"sinks,omitempty"`) to `SourceConfig`; leave `Sink` and `Peer` in place. +3. Add a resolver method (e.g. `ResolvedSinks()`) that returns `Sinks` when non-empty, else a one-element list synthesized from `Sink`+`Peer`. Every consumer calls this rather than reading fields directly, so there is one place the legacy fallback lives. When both `Sinks` and legacy `Sink.URL` are empty, return an empty list (no delivery) so a mis-written config fails visibly rather than POSTing to an empty URL. +4. **Writing is via string template, not `yaml.Marshal`.** No code marshals `SourceConfig` today; configs are written by `renderSourceYAML` (`internal/cli/wizard.go`). Extend that renderer to emit a `sinks:` block for multi-sink configs and keep the single-sink render for legacy. Do not introduce a `yaml.Marshal` path in this unit: `Sink SinkRef` is tagged `yaml:"sink"` without `omitempty` and yaml.v3 does not omit zero-value structs, so marshaling a multi-sink-only config would emit a stray empty `sink:` block. (If a future unit does add marshaling, change `Sink` to `*SinkRef` first.) + +**Patterns to follow:** the existing `omitempty` refs in this file (`Cmux`, `CDP`, `LiveCDP`) and their doc-comment style; `renderSourceYAML` in `internal/cli/wizard.go` for the write path. + +**Test scenarios:** +- A `source.yaml` with only legacy `sink:`/`peer:` resolves to a one-element sink list with the same URL and peer. +- A `source.yaml` with a two-element `sinks:` list resolves to both, and the legacy scalars are ignored when `sinks` is present. +- A config with neither `sinks` nor legacy `sink.url` resolves to an empty list (no delivery), not a one-element empty-URL sink. +- Decoding a hand-written multi-sink `sinks:` block yields the expected per-entry `url` and `peer` (decode, not marshal round-trip). + +**Verification:** config tests pass; a hand-written legacy config and a multi-sink config both decode to the expected resolved list. + +### U2. Fan-out push: seal and POST per sink, isolate failures + +**Goal:** one push reads and filters once, then delivers to every resolved sink, continuing past a failed sink. + +**Requirements:** R3, R4, R7. + +**Dependencies:** U1. + +**Files:** +- `internal/cli/source.go` — `runSource`, `pushWithFreshBlocklist`, `pushOnce`, `recordSourcePushResult`. +- `internal/cli/source_test.go` — fan-out and failure-isolation coverage. + +**Approach:** +1. In `runSource`, resolve the sink list via `ResolvedSinks()`. Do **not** resolve secrets up front — `resolveTransportSecret` hard-errors on a missing key and would abort all sinks (per KTD2, this defeats R4). +2. In `pushOnce`, build the plaintext payload once (cookies after blocklist/DBSC filtering plus the secrets-bus payload — unchanged), then loop the sinks. Per sink: resolve `resolveTransportSecret(common.ConfigDir, sink.Peer, cfg.Security.SharedSecret)` (note the third arg is the legacy shared-secret string), `SealWithSecret(payload, secret)`, POST to `sink.URL` bounded by the per-sink timeout, capture a per-sink result. +3. Isolate errors: a secret-resolution, seal, or POST failure records that sink as failed and continues to the next sink. Never fall through to the shared secret on a missing per-sink key (KTD2). Aggregate into a whole/partial/total-failure outcome for the return value and exit status. +4. In `--once`, scale (or drop) the outer `context.WithTimeout` wrapper per KTD3 so a slow first sink cannot cancel later sinks. +5. Keep the existing stderr reply line per sink (`posted N cookies, sink replied: ...`) prefixed with the sink so three-sink output stays readable, and keep the single `secrets-bus: shipping N cli(s)` line (secrets are computed once). +6. `--watch` calls the same `pushOnce`, so fan-out on every cookie/secrets write falls out of the loop with no watcher change. + +**Execution note:** start with a failing test that asserts a two-sink push where the first sink errors still delivers to the second and returns a partial-success status. + +**Patterns to follow:** existing `pushOnce` sealing and POST with the `SyncClient` timeout profile; existing `recordSourcePushResult` shape. + +**Test scenarios:** +- Two healthy sinks: one push seals twice with the two peer keys and POSTs to both URLs; both recorded ok. +- First sink POST times out, second healthy: second still receives the payload; outcome is partial success; first sink's error recorded. +- First sink's peer key is missing: that sink is recorded failed and the loop continues to the healthy second sink; the missing-key sink is NOT sealed under `Security.SharedSecret`. +- `--once` with a first sink that runs to its full per-sink timeout: later healthy sinks still get their own full timeout window and deliver (guards the scaled outer deadline from KTD3). +- All sinks fail: push returns total failure with each sink's error recorded. +- Single legacy sink (resolved one-element list): behavior and stderr identical to today. +- Cookies are read and filtered exactly once regardless of sink count (assert the read/filter path runs once for a three-sink config). + +**Verification:** source tests pass; a manual `source --once` against two reachable local listeners delivers to both, and killing one listener leaves the other delivered with a partial-success exit. + +### U3. Per-sink source state + +**Goal:** push results are tracked per sink so status and doctor can report each target. + +**Requirements:** R6. + +**Dependencies:** U1. + +**Files:** +- `internal/state/state.go` — `SourceState` per-sink records. +- `internal/state/state_test.go` — serialization and back-compat coverage. + +**Approach:** +1. Add a per-sink record type keyed by peer hostname (per KTD4), carrying the current URL, `LastPush`, `LastPushCount`, `TotalPushes`, `TotalFailures`, `LastError`, `LastErrorAt`, and hold a list/map of them on `SourceState`. +2. Keep the top-level DBSC summary fields (`LastDBSCWarned/Skipped/Sample`) since the filtered set is shared across sinks. +3. Tolerate an old `sink_url`-shaped state file on first read after upgrade: `state.LoadSource` does a plain `json.Unmarshal` with no versioning, so keep the legacy `SinkURL` field decodable and migrate it into a one-element per-sink record rather than erroring. + +**Patterns to follow:** existing `SourceState` JSON tags and the DBSC field doc comments. + +**Test scenarios:** +- Recording two sinks' results yields two per-sink records with independent counters. +- A pre-upgrade state JSON with `sink_url` loads into a single per-sink record without error. +- A total-failure push increments the failing sink's `TotalFailures` and sets its `LastError`, leaving a healthy sink's counters untouched. + +**Verification:** state tests pass; the state file after a two-sink push shows two records. + +### U4. Additive wizard pairing for a new sink + +**Goal:** grow an existing source config by pairing another peer and appending it to `sinks`, without overwriting existing sinks. + +**Requirements:** R5. + +**Dependencies:** U1. + +**Files:** +- `internal/cli/wizard.go` — `runWizardInstall`, `wizardInstallSource`, `renderSourceYAML`, `beginSourcePairing` / the pairing-info writer, and the `guardConfigPeerMismatch` path. +- `internal/cli/wizard_test.go` — additive-path coverage. + +**Approach:** +1. Add an additive mode (an `--add-sink` flag, or make `--peer` recognize that a source config already exists and append rather than overwrite). Explicitly not the current `--force`, which overwrites `source.yaml`. +2. In additive mode, skip the `guardConfigPeerMismatch` overwrite guard: load the existing config, pair the new peer, then append a `SinkTarget{URL, Peer}` to `Sinks`, migrating a legacy single-sink config to a two-element list on first add. +3. **File the new key under the operator-supplied `--peer` name via `beginSourcePairing` / the pairing-info writer — not `runPairAsSource`.** `runPairAsSource` stores the key under `res.RemotePeer` (the sink's announced `os.Hostname`, e.g. `e2b.local`), while `SinkTarget.Peer` and the fan-out lookup (`resolveTransportSecret(..., sink.Peer, ...)`) use the operator name (e.g. `instinct`). Filing under the announced name would leave the daemon looking up the wrong `keys/.json` at sync time and reporting connection-refused. Record the announced hostname separately (as the wizard already does), so `keys/.json` matches the name written into `Sinks`. This is the exact "announced as e2b.local, stored as instinct" behavior the wizard already handles for the single-sink case. +4. Render `sinks:` via `renderSourceYAML` when writing a multi-sink config; keep the legacy single-sink render for the first-time single install. +5. Leave the source LaunchAgent unchanged — it already runs `source --watch`, which now fans out via U2. + +**Execution note:** this unit changes pairing/config-write flow; add a test that an add-sink against a legacy single-sink config yields a two-element `sinks` list with `keys/.json` present under the operator peer name, before wiring the flag. + +**Patterns to follow:** existing `wizardInstallSource` write path, `beginSourcePairing` / pairing-info writer key-filing, `keystore.Path`, and `renderSourceYAML`. + +**Test scenarios:** +- Add-sink against a legacy single-sink config produces a two-element `sinks` list and a second `keys/.json` named by the operator `--peer`, not the announced hostname. +- Add-sink against an existing multi-sink config appends a third entry and leaves the first two intact. +- Add-sink with a peer that is already present is a no-op (or clear error), not a duplicate entry. +- A first-time single install still writes the legacy single-sink shape and pairs one peer. + +**Verification:** wizard tests pass; adding a sink to a live single-sink config leaves the original sink and key in place and the daemon then pushes to both. + +### U5. Per-sink reporting in status and doctor + +**Goal:** `status` and `doctor` show each sink's URL, last push, last error, and reachability. + +**Requirements:** R6. + +**Dependencies:** U3. + +**Files:** +- `internal/cli/status.go` — render per-sink source state. +- `internal/cli/doctor.go` — per-sink source-state and reachability checks. +- `internal/cli/doctor_test.go` / `status_test.go` — per-sink output coverage. + +**Approach:** +1. `status`: replace the single sink URL/last-push block with one block per configured sink from the per-sink state. +2. `doctor`: the existing "Source state" check reports per sink (last push age, last error); keep it green/warn per sink. Reachability, if probed, is reported per sink URL. +3. Keep output stable and readable for the common single-sink case (one block reads like today). + +**Patterns to follow:** existing `doctor` check list and the `[OK]/[WARN]` line format; existing `status` config/state rendering. + +**Test scenarios:** +- `status` with two sinks prints two sink blocks with independent last-push values. +- `doctor` with one stale sink and one fresh sink reports WARN for the stale one and OK for the fresh one. +- Single-sink config prints a single block matching today's shape. + +**Verification:** status/doctor tests pass; `doctor` on a two-sink install reports both sinks. + +### U6. Docs: README, CHANGELOG, example config + +**Goal:** the multi-sink shape and the add-sink flow are documented, and the DBSC per-sink caveat is stated. + +**Requirements:** R1, R5. + +**Dependencies:** U1, U4. + +**Files:** +- `README.md` — multi-sink `sinks:` config, the add-sink command, and the per-sink DBSC note. +- `CHANGELOG.md` — the multi-sink entry. +- `examples/` — a multi-sink `source.yaml` example. + +**Approach:** document the `sinks:` list, the backward-compatible single-sink form, the add-sink flow from U4, and that device-bound cookies remain per-sink. Mirror the existing README section style. + +**Test expectation:** none -- docs and example config, no behavioral change. + +**Verification:** README renders; the example config decodes under U1's loader (a decode test over `examples/*.yaml` if one exists, otherwise manual). + +--- + +## Scope Boundaries + +**In scope:** the source side — config schema, fan-out push, per-sink state, additive wizard pairing, status/doctor reporting, docs. + +**Out of scope:** the sink side (no sink behavior change — each sink still receives one sealed payload exactly as today), the transport/crypto (seal format and key derivation unchanged), DBSC handling (unchanged, and per-sink by nature), and the cmux local loop (`cmux-sync` is same-machine and independent of the push). + +### Deferred to Follow-Up Work + +- **Concurrent fan-out.** Sequential POSTs with per-sink timeouts are the initial approach (KTD3). Parallelizing the per-sink POSTs is a later optimization if the sink count grows past a handful. +- **A `remove-sink` / list-sinks management command.** This plan adds sinks; pruning one is a hand-edit for now. Treat this as a security control, not just convenience (see Open Questions): the interim manual revoke for a compromised or retired sink is to remove its `SinkTarget` entry **and** delete `keys/.json`, so it stops receiving the credential stream on the next watch cycle. Until `remove-sink` exists, that manual procedure is the revocation path. +- **Per-sink blocklist or domain filtering.** The blocklist stays global to the source; per-sink cookie narrowing is not in scope. + +--- + +## System-Wide Impact + +- **Existing single-sink users:** unaffected. Legacy configs load as a one-element list (KTD1) and the stderr and state shapes for one sink match today. +- **State file:** the on-disk `SourceState` shape changes; U3 migrates an old `sink_url` state on first read rather than erroring. +- **Operators:** `status`/`doctor` output grows a block per sink; single-sink output stays familiar. + +--- + +## Open Questions + +These are trust-model and delivery-behavior forks that need Matt's call; they do not block starting U1-U6 but should be settled before shipping. + +- **OQ1 — Sink trust tiering.** Every sink receives the identical full cookie and secret payload, so the weakest sink's compromise exposes the whole credential set. Are all three sinks (instinct/e2b, Mac mini, grok-bot) accepted as equally trusted with the full payload, or does a lower-trust sink like grok-bot warrant a scoped payload (per-sink domain filtering, deferred here) before multi-sink ships? Default assumption if unanswered: all sinks equally trusted, stated as such. +- **OQ2 — Revocation as a first-class control.** Should `remove-sink` be pulled into this PR rather than deferred, given that without it a compromised or decommissioned sink keeps receiving fresh credentials until a manual key delete? Default: keep it deferred with the documented manual revoke procedure above. +- **OQ3 — Chronically-dead sink under `--watch`.** A sink that is off but reachable-then-timing-out costs its full 5-minute timeout serially on every cookie change, degrading freshness to healthy sinks. This is the exact three-machines-one-often-off case. Options: a short fast-fail connect timeout, per-sink backoff that skips a recently-failed sink, or bring concurrency (currently deferred) forward. Default: fast-fail connect timeout, backoff deferred. + +--- + +## Verification Contract + +- Package tests pass for `internal/config`, `internal/cli` (source, wizard, status, doctor), and `internal/state`. +- A legacy single-sink `source.yaml` produces byte-for-byte-equivalent push behavior and stderr for its one sink. +- A two-sink `source --once` delivers to both reachable listeners; with one listener down, the other is still delivered and the push reports partial success. +- In `--once`, a first sink that runs to its full per-sink timeout does not cancel later healthy sinks (the scaled outer deadline holds). +- A sink whose per-sink key is missing is isolated as a failure and is never sealed under the fleet-wide shared secret. +- `doctor` on a two-sink install reports both sinks with independent last-push/last-error state. +- Adding a sink via the wizard to a live single-sink config leaves the original sink and key intact, files the new key as `keys/.json` (not the announced hostname), and the watcher then pushes to both. + +## Definition of Done + +- R1-R7 met. +- All units landed with their test scenarios covered. +- Legacy single-sink configs verified unchanged. +- README, CHANGELOG, and an example multi-sink config updated. +- No sink-side, transport, or DBSC behavior changed. From e37749f670167d207b3892420fdd74d5b83e655a Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Thu, 10 Sep 2026 05:56:20 -0700 Subject: [PATCH 7/7] style: satisfy golangci-lint (max builtin, fmt.Fprintf) Use the max builtin for the --once sink-count floor and fmt.Fprintf into the strings.Builder in renderSourceYAMLSinks (QF1012), clearing the go-lint CI failures on the multi-sink branch. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PFRsAqcEwq1jZnRk4zFQyy --- internal/cli/source.go | 5 +---- internal/cli/wizard.go | 16 ++++++++-------- 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/internal/cli/source.go b/internal/cli/source.go index e8dba95..1203a8b 100644 --- a/internal/cli/source.go +++ b/internal/cli/source.go @@ -150,10 +150,7 @@ func runSource(cmd *cobra.Command, args []string) error { // timeout must not exhaust the shared deadline and starve later // healthy sinks (which would defeat per-sink isolation). One // per-sink ClientTimeout per sink, plus slack. - nSinks := len(sinks) - if nSinks < 1 { - nSinks = 1 - } + nSinks := max(len(sinks), 1) perSink := httpserver.Defaults(httpserver.SyncClient).ClientTimeout ctx, cancel := context.WithTimeout(cmd.Context(), time.Duration(nSinks)*perSink+30*time.Second) defer cancel() diff --git a/internal/cli/wizard.go b/internal/cli/wizard.go index 6e834f4..a24570a 100644 --- a/internal/cli/wizard.go +++ b/internal/cli/wizard.go @@ -762,9 +762,9 @@ func renderSourceYAMLSinks(cfg *config.SourceConfig, sinks []config.SinkTarget) var b strings.Builder b.WriteString("sinks:\n") for _, s := range sinks { - b.WriteString(fmt.Sprintf(" - url: %s\n", s.URL)) + fmt.Fprintf(&b, " - url: %s\n", s.URL) if s.Peer != "" { - b.WriteString(fmt.Sprintf(" peer: %s\n", s.Peer)) + fmt.Fprintf(&b, " peer: %s\n", s.Peer) } } dbPath := cfg.Chrome.DBPath @@ -772,25 +772,25 @@ func renderSourceYAMLSinks(cfg *config.SourceConfig, sinks []config.SinkTarget) dbPath = "~/Library/Application Support/Google/Chrome/Default/Cookies" } b.WriteString("chrome:\n") - b.WriteString(fmt.Sprintf(" db_path: %s\n", dbPath)) + fmt.Fprintf(&b, " db_path: %s\n", dbPath) if cfg.Browser.Name != "" || cfg.Browser.Profile != "" { b.WriteString("browser:\n") if cfg.Browser.Name != "" { - b.WriteString(fmt.Sprintf(" name: %s\n", cfg.Browser.Name)) + fmt.Fprintf(&b, " name: %s\n", cfg.Browser.Name) } if cfg.Browser.Profile != "" { - b.WriteString(fmt.Sprintf(" profile: %s\n", cfg.Browser.Profile)) + fmt.Fprintf(&b, " profile: %s\n", cfg.Browser.Profile) } } if cfg.Security.SharedSecret != "" { b.WriteString("security:\n") - b.WriteString(fmt.Sprintf(" shared_secret: %s\n", cfg.Security.SharedSecret)) + fmt.Fprintf(&b, " shared_secret: %s\n", cfg.Security.SharedSecret) } if cfg.Cmux.Enabled { b.WriteString("cmux:\n") - b.WriteString(fmt.Sprintf(" enabled: %v\n", cfg.Cmux.Enabled)) + fmt.Fprintf(&b, " enabled: %v\n", cfg.Cmux.Enabled) if cfg.Cmux.CmuxPath != "" { - b.WriteString(fmt.Sprintf(" cmux_path: %s\n", cfg.Cmux.CmuxPath)) + fmt.Fprintf(&b, " cmux_path: %s\n", cfg.Cmux.CmuxPath) } } return b.String()