From 2ce714ac4d6b8f1ef2203a14bdea8b1b6f1d7b6d Mon Sep 17 00:00:00 2001 From: Snow Lee Date: Sat, 11 Jul 2026 14:49:02 -0700 Subject: [PATCH] =?UTF-8?q?feat(sync):=20agent=20read=20reporting=20?= =?UTF-8?q?=E2=80=94=20read-log,=20spool,=20hub=20report,=20hook=20matcher?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of the read heatmap: `bdrive read-log` parses any platform's hook event JSON from stdin and queues in-project file reads (mount-relative, ignore/include-filtered) in a per-volume spool — no network on the hook path. The sync cycle drains the spool best-effort to the hub's new POST /api/p//reads via the remote ReadReporter capability (https backend only); a failed report retries next cycle and never fails or offlines the cycle. `bdrive hooks install` now registers a third hook per platform on its read-tool matcher (claude Read, codex read_file best-effort, gemini read_file|read_many_files, hermes read_file), each idempotent on its own marker so sync-only configs upgrade in place. Agent reads land as agent traffic in the heat view, actor = device id. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01P5cxPQdSGJnjXCYY9GeWXt --- cmd/bdrive/main.go | 1 + cmd/bdrive/readlog.go | 131 +++++++++++++++++++++++++ cmd/bdrive/readlog_test.go | 93 ++++++++++++++++++ internal/agenthooks/agenthooks.go | 89 ++++++++++++----- internal/agenthooks/agenthooks_test.go | 92 ++++++++++++++++- internal/remote/http.go | 24 +++++ internal/remote/remote.go | 14 +++ internal/store/reads.go | 107 ++++++++++++++++++++ internal/store/reads_test.go | 111 +++++++++++++++++++++ internal/syncer/ignore.go | 8 ++ internal/syncer/reads_flow_test.go | 101 +++++++++++++++++++ internal/syncer/syncer.go | 15 +++ 12 files changed, 758 insertions(+), 28 deletions(-) create mode 100644 cmd/bdrive/readlog.go create mode 100644 cmd/bdrive/readlog_test.go create mode 100644 internal/store/reads.go create mode 100644 internal/store/reads_test.go create mode 100644 internal/syncer/reads_flow_test.go diff --git a/cmd/bdrive/main.go b/cmd/bdrive/main.go index 52d5c65..3624ada 100644 --- a/cmd/bdrive/main.go +++ b/cmd/bdrive/main.go @@ -36,6 +36,7 @@ everything keeps working offline; changes sync when the remote is reachable.`, shareCmd(), stopCmd(), syncCmd(), + readLogCmd(), hooksCmd(), statusCmd(), logCmd(), diff --git a/cmd/bdrive/readlog.go b/cmd/bdrive/readlog.go new file mode 100644 index 0000000..b1ca6c8 --- /dev/null +++ b/cmd/bdrive/readlog.go @@ -0,0 +1,131 @@ +package main + +import ( + "encoding/json" + "io" + "path/filepath" + "strings" + + "github.com/spf13/cobra" + + "github.com/runbear-io/beardrive/internal/config" + "github.com/runbear-io/beardrive/internal/store" + "github.com/runbear-io/beardrive/internal/syncer" +) + +// read-log is the agent read hook's command: it runs after every file-read +// tool call, so it must be a fast, silent no-op in every case that isn't +// "a synced file was just read" — no network, no locking, one appended line. +func readLogCmd() *cobra.Command { + return &cobra.Command{ + Use: "read-log [folder]", + Short: "Record agent file reads from a hook event (JSON on stdin)", + Long: `Record which project files an agent just read, from the hook event JSON +piped on stdin (any platform's PostToolUse-style payload: file paths are +found wherever they appear in the event). + +Reads are queued locally in the volume store and drained to the hub on the +next sync, where they show up as agent traffic in the project's read heat. +Registered automatically by "bdrive hooks install"; there is rarely a reason +to run it by hand.`, + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + folder, err := absFolder(args) + if err != nil { + return nil + } + proj, found, err := config.ResolveMount(folder) + if err != nil || !found { + return nil // not a beardrive project: fast no-op + } + data, _ := io.ReadAll(io.LimitReader(cmd.InOrStdin(), 1<<20)) + paths := extractEventPaths(data) + if len(paths) == 0 { + return nil + } + filter, err := syncer.LoadFilter(folder, proj.Include) + if err != nil { + return nil + } + vdir, err := config.VolumeDir(proj.ID) + if err != nil { + return nil + } + st, err := store.Open(vdir) + if err != nil { + return nil + } + for _, p := range paths { + abs := p + if !filepath.IsAbs(abs) { + abs = filepath.Join(folder, p) + } + rel, err := filepath.Rel(folder, abs) + if err != nil || rel == "." || strings.HasPrefix(rel, "..") { + continue // outside the mount + } + rel = filepath.ToSlash(rel) + if filter.Skip(rel) { + continue // not part of the project (ignore/include rules) + } + st.LogRead(rel) // best-effort; the hook must never fail the turn + } + return nil + }, + } +} + +// Keys that carry file paths in the hook payloads of the supported agent +// platforms (Claude Code tool_input.file_path, Gemini/Hermes read_file +// path/absolute_path, multi-read paths arrays). +var ( + eventPathKeys = map[string]bool{"file_path": true, "absolute_path": true, "path": true, "notebook_path": true} + eventPathListKeys = map[string]bool{"paths": true, "file_paths": true} +) + +// extractEventPaths pulls candidate file paths out of an arbitrary hook +// event JSON. The hook only fires on read-tool matchers, so any path-shaped +// field is a read; non-project paths are filtered by the caller. +func extractEventPaths(data []byte) []string { + var root any + if json.Unmarshal(data, &root) != nil { + return nil + } + seen := map[string]bool{} + var out []string + add := func(s string) { + if s != "" && !seen[s] { + seen[s] = true + out = append(out, s) + } + } + var walk func(v any) + walk = func(v any) { + switch t := v.(type) { + case map[string]any: + for k, val := range t { + switch { + case eventPathKeys[k]: + if s, ok := val.(string); ok { + add(s) + } + case eventPathListKeys[k]: + if arr, ok := val.([]any); ok { + for _, it := range arr { + if s, ok := it.(string); ok { + add(s) + } + } + } + } + walk(val) + } + case []any: + for _, it := range t { + walk(it) + } + } + } + walk(root) + return out +} diff --git a/cmd/bdrive/readlog_test.go b/cmd/bdrive/readlog_test.go new file mode 100644 index 0000000..ff0c040 --- /dev/null +++ b/cmd/bdrive/readlog_test.go @@ -0,0 +1,93 @@ +package main + +import ( + "bytes" + "path/filepath" + "strings" + "testing" + + "github.com/runbear-io/beardrive/internal/config" + "github.com/runbear-io/beardrive/internal/store" +) + +// Real hook payload shapes from the supported platforms — read-log must find +// the file paths wherever each platform puts them. +func TestExtractEventPaths(t *testing.T) { + cases := map[string]struct { + payload string + want []string + }{ + "claude read": { + `{"session_id":"abc","hook_event_name":"PostToolUse","tool_name":"Read", + "tool_input":{"file_path":"/proj/wiki/a.md"},"tool_response":{"type":"text"}}`, + []string{"/proj/wiki/a.md"}, + }, + "gemini read_many_files": { + `{"session_id":"g1","tool":{"name":"read_many_files","args":{"paths":["wiki/a.md","wiki/b.md"]}}}`, + []string{"wiki/a.md", "wiki/b.md"}, + }, + "gemini read_file absolute": { + `{"session_id":"g2","tool":{"name":"read_file","args":{"absolute_path":"/proj/notes.md"}}}`, + []string{"/proj/notes.md"}, + }, + "hermes read_file": { + `{"hook_event_name":"post_tool_call","tool_name":"read_file","tool_args":{"path":"docs/x.md"}}`, + []string{"docs/x.md"}, + }, + "duplicates collapse": { + `{"tool_input":{"file_path":"a.md"},"extra":{"file_path":"a.md"}}`, + []string{"a.md"}, + }, + "no paths": { + `{"session_id":"abc","prompt":"hello"}`, + nil, + }, + "not json": { + `plain text`, + nil, + }, + } + for name, c := range cases { + got := extractEventPaths([]byte(c.payload)) + if strings.Join(got, ",") != strings.Join(c.want, ",") { + t.Errorf("%s: paths = %v, want %v", name, got, c.want) + } + } +} + +// End to end through the cobra command: a claude-style event lands in the +// mount's read spool, filtered to project-relative synced paths. +func TestReadLogCommand(t *testing.T) { + t.Setenv("BDRIVE_HOME", t.TempDir()) + folder := t.TempDir() + folder, _ = filepath.EvalSymlinks(folder) + proj, err := config.SaveProject(folder, config.Project{Volume: "wiki"}) + if err != nil { + t.Fatal(err) + } + + c := readLogCmd() + c.SetIn(bytes.NewReader([]byte(`{"session_id":"abc","tool_name":"Read", + "tool_input":{"file_path":"` + folder + `/wiki/a.md"}, + "other":{"file_path":"/somewhere/else/entirely.md"}}`))) + c.SetArgs([]string{folder}) + if err := c.Execute(); err != nil { + t.Fatal(err) + } + + vdir, err := config.VolumeDir(proj.ID) + if err != nil { + t.Fatal(err) + } + st, err := store.Open(vdir) + if err != nil { + t.Fatal(err) + } + evs, err := st.PendingReads() + if err != nil { + t.Fatal(err) + } + if len(evs) != 1 || evs[0].Path != "wiki/a.md" { + t.Fatalf("spool = %+v, want just the in-project read, mount-relative", evs) + } +} diff --git a/internal/agenthooks/agenthooks.go b/internal/agenthooks/agenthooks.go index f6200bd..2cf08a9 100644 --- a/internal/agenthooks/agenthooks.go +++ b/internal/agenthooks/agenthooks.go @@ -13,7 +13,10 @@ // // The hook syncs the project and stamps changes with " session " // (see `bdrive sync --note`), so hub history links every change to the agent -// session that made it. Hooks are fast no-ops outside bdrive projects. +// session that made it. A third hook on each platform's read-tool matcher +// runs `bdrive read-log`, queueing agent file reads for the hub's read +// heatmap (drained on the next sync — the hook itself never touches the +// network). Hooks are fast no-ops outside bdrive projects. package agenthooks import ( @@ -28,8 +31,14 @@ import ( "github.com/runbear-io/beardrive/internal/store" ) -// marker identifies our hooks inside a config, for idempotency and status. -const marker = "bdrive sync" +// Markers identify our hooks inside a config, for idempotency and status. +// The sync and read hooks are separate groups (different matchers), so each +// carries its own marker — re-running install on a config that predates the +// read hook adds just the missing group. +const ( + marker = "bdrive sync" + readMarker = "bdrive read-log" +) // Agent names, in the order they are reported. var Agents = []string{"claude", "codex", "gemini", "hermes"} @@ -53,6 +62,15 @@ func hookCommand(label string) string { `else bdrive sync . >/dev/null 2>&1 || true; fi'` } +// readHookCommand queues agent file reads for the hub's read heatmap: +// `bdrive read-log` parses the hook's stdin JSON itself and only appends to +// a local spool, so this stays cheap enough to run on every read-tool call. +func readHookCommand() string { + return `sh -c '` + + `cd "${CLAUDE_PROJECT_DIR:-.}" && [ -d .bdrive ] && command -v bdrive >/dev/null || exit 0; ` + + `bdrive read-log . >/dev/null 2>&1 || true'` +} + type platform struct { label string // session-note label projectDir string // presence of this dir (project or home) = detected @@ -67,15 +85,17 @@ var platforms = map[string]platform{ projectDir: ".claude", install: func(folder string) (string, bool, error) { return mergeJSONHooks(filepath.Join(folder, ".claude", "settings.json"), - "UserPromptSubmit", "PostToolUse", "Write|Edit|MultiEdit", "claude-code", 30, true) + "UserPromptSubmit", "PostToolUse", "Write|Edit|MultiEdit", "Read", "claude-code", 30, true) }, }, "codex": { label: "codex", projectDir: ".codex", install: func(folder string) (string, bool, error) { + // Codex reads mostly happen through shell commands, so the + // read_file matcher is best-effort coverage. return mergeJSONHooks(filepath.Join(folder, ".codex", "hooks.json"), - "UserPromptSubmit", "PostToolUse", "apply_patch", "codex", 30, false) + "UserPromptSubmit", "PostToolUse", "apply_patch", "read_file", "codex", 30, false) }, note: "run /hooks inside Codex once to trust the project's .codex layer", }, @@ -85,7 +105,7 @@ var platforms = map[string]platform{ install: func(folder string) (string, bool, error) { // Gemini uses its own event names and millisecond timeouts. return mergeJSONHooks(filepath.Join(folder, ".gemini", "settings.json"), - "BeforeAgent", "AfterTool", "write_file|replace|edit", "gemini", 30000, false) + "BeforeAgent", "AfterTool", "write_file|replace|edit", "read_file|read_many_files", "gemini", 30000, false) }, }, "hermes": { @@ -159,10 +179,12 @@ func Install(folder string, agents []string) ([]Result, error) { return out, nil } -// mergeJSONHooks adds the pull + push hook pair to a Claude-style hooks JSON -// file (Claude, Codex, and Gemini all use this shape: hooks. is an -// array of {matcher?, hooks: [{type: "command", ...}]} groups). -func mergeJSONHooks(path, pullEvent, pushEvent, pushMatcher, label string, timeout int, async bool) (string, bool, error) { +// mergeJSONHooks adds the pull + push + read hook trio to a Claude-style +// hooks JSON file (Claude, Codex, and Gemini all use this shape: +// hooks. is an array of {matcher?, hooks: [{type: "command", ...}]} +// groups). Push and read share the tool-use event under different matchers, +// each idempotent on its own marker. +func mergeJSONHooks(path, pullEvent, pushEvent, pushMatcher, readMatcher, label string, timeout int, async bool) (string, bool, error) { root := map[string]any{} if data, err := os.ReadFile(path); err == nil { if err := json.Unmarshal(data, &root); err != nil { @@ -182,18 +204,29 @@ func mergeJSONHooks(path, pullEvent, pushEvent, pushMatcher, label string, timeo "statusMessage": "beardrive: pulling latest files", }}} pushHook := map[string]any{"type": "command", "command": cmd, "timeout": timeout} + readHook := map[string]any{"type": "command", "command": readHookCommand(), "timeout": timeout} if async { pushHook["async"] = true + readHook["async"] = true } push := map[string]any{"matcher": pushMatcher, "hooks": []any{pushHook}} + read := map[string]any{"matcher": readMatcher, "hooks": []any{readHook}} changed := false - for event, group := range map[string]any{pullEvent: pull, pushEvent: push} { - arr, _ := hooks[event].([]any) - if containsMarker(arr) { + for _, g := range []struct { + event string + group map[string]any + marker string + }{ + {pullEvent, pull, marker}, + {pushEvent, push, marker}, + {pushEvent, read, readMarker}, + } { + arr, _ := hooks[g.event].([]any) + if containsMarker(arr, g.marker) { continue } - hooks[event] = append(arr, group) + hooks[g.event] = append(arr, g.group) changed = true } if !changed { @@ -222,17 +255,22 @@ func installHermes(string) (string, bool, error) { root["hooks"] = hooks } cmd := hookCommand("hermes") - groups := map[string]any{ - "pre_llm_call": map[string]any{"command": cmd, "timeout": 30}, - "post_tool_call": map[string]any{"matcher": "write_file|patch", "command": cmd, "timeout": 30}, + groups := []struct { + event string + group map[string]any + marker string + }{ + {"pre_llm_call", map[string]any{"command": cmd, "timeout": 30}, marker}, + {"post_tool_call", map[string]any{"matcher": "write_file|patch", "command": cmd, "timeout": 30}, marker}, + {"post_tool_call", map[string]any{"matcher": "read_file", "command": readHookCommand(), "timeout": 30}, readMarker}, } changed := false - for event, group := range groups { - arr, _ := hooks[event].([]any) - if containsMarker(arr) { + for _, g := range groups { + arr, _ := hooks[g.event].([]any) + if containsMarker(arr, g.marker) { continue } - hooks[event] = append(arr, group) + hooks[g.event] = append(arr, g.group) changed = true } if !changed { @@ -243,11 +281,12 @@ func installHermes(string) (string, bool, error) { }) } -// containsMarker reports whether a hook array already holds one of ours. -// Serializing sidesteps walking every platform's nesting by hand. -func containsMarker(v any) bool { +// containsMarker reports whether a hook array already holds the hook the +// marker identifies. Serializing sidesteps walking every platform's nesting +// by hand. +func containsMarker(v any, m string) bool { data, err := json.Marshal(v) - return err == nil && strings.Contains(string(data), marker) + return err == nil && strings.Contains(string(data), m) } func writeConfig(path string, marshal func() ([]byte, error)) error { diff --git a/internal/agenthooks/agenthooks_test.go b/internal/agenthooks/agenthooks_test.go index 30b1a54..d66d82a 100644 --- a/internal/agenthooks/agenthooks_test.go +++ b/internal/agenthooks/agenthooks_test.go @@ -75,6 +75,10 @@ func TestInstallJSONPlatforms(t *testing.T) { if !strings.Contains(string(raw), `"async":true`) { t.Fatal("claude push hook should be async") } + // …and the read hook, on its own matcher. + if !strings.Contains(string(raw), "bdrive read-log") || !strings.Contains(string(raw), `"matcher":"Read"`) { + t.Fatalf("claude read hook missing: %s", raw) + } // Codex: same schema, its own label and matcher, no async field. cx, _ := json.Marshal(readJSON(t, filepath.Join(folder, ".codex", "hooks.json"))) @@ -84,10 +88,13 @@ func TestInstallJSONPlatforms(t *testing.T) { if strings.Contains(string(cx), "async") { t.Fatal("codex should not get the claude-only async field") } + if !strings.Contains(string(cx), "bdrive read-log") || !strings.Contains(string(cx), `"matcher":"read_file"`) { + t.Fatalf("codex read hook missing: %s", cx) + } // Gemini: its own event names and ms timeout. gm, _ := json.Marshal(readJSON(t, filepath.Join(folder, ".gemini", "settings.json"))) - for _, want := range []string{"BeforeAgent", "AfterTool", "gemini session $s", "30000"} { + for _, want := range []string{"BeforeAgent", "AfterTool", "gemini session $s", "30000", "bdrive read-log", "read_file|read_many_files"} { if !strings.Contains(string(gm), want) { t.Fatalf("gemini hooks missing %q: %s", want, gm) } @@ -114,8 +121,8 @@ func TestInstallIdempotentAndPreserving(t *testing.T) { if _, ok := cfg["permissions"]; !ok { t.Fatal("merge dropped unrelated settings keys") } - if got := len(cfg["hooks"].(map[string]any)["PostToolUse"].([]any)); got != 2 { - t.Fatalf("PostToolUse groups = %d, want user's + ours", got) + if got := len(cfg["hooks"].(map[string]any)["PostToolUse"].([]any)); got != 3 { + t.Fatalf("PostToolUse groups = %d, want user's + our push + our read", got) } // Second install: no change, byte-identical file. @@ -133,6 +140,38 @@ func TestInstallIdempotentAndPreserving(t *testing.T) { } } +// A config from before the read hook existed (sync hooks only) gains just +// the read group on re-install — the sync hooks are not duplicated. +func TestInstallUpgradesSyncOnlyConfig(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + folder := t.TempDir() + old := `{"hooks":{ + "UserPromptSubmit":[{"hooks":[{"type":"command","command":"sh -c 'bdrive sync .'"}]}], + "PostToolUse":[{"matcher":"Write|Edit|MultiEdit","hooks":[{"type":"command","command":"sh -c 'bdrive sync .'"}]}]}}` + os.MkdirAll(filepath.Join(folder, ".claude"), 0o755) + os.WriteFile(filepath.Join(folder, ".claude", "settings.json"), []byte(old), 0o644) + + results, err := Install(folder, []string{"claude"}) + if err != nil { + t.Fatal(err) + } + if !results[0].Changed { + t.Fatal("upgrade install reported unchanged") + } + cfg := readJSON(t, filepath.Join(folder, ".claude", "settings.json")) + hooks := cfg["hooks"].(map[string]any) + if got := len(hooks["UserPromptSubmit"].([]any)); got != 1 { + t.Fatalf("UserPromptSubmit groups = %d, want the old one only", got) + } + if got := len(hooks["PostToolUse"].([]any)); got != 2 { + t.Fatalf("PostToolUse groups = %d, want old push + new read", got) + } + raw, _ := json.Marshal(cfg) + if !strings.Contains(string(raw), "bdrive read-log") { + t.Fatal("upgrade did not add the read hook") + } +} + func TestInstallHermesYAML(t *testing.T) { home := t.TempDir() t.Setenv("HOME", home) @@ -162,9 +201,15 @@ func TestInstallHermesYAML(t *testing.T) { t.Fatalf("hermes missing %s", ev) } } + if got := len(hooks["post_tool_call"].([]any)); got != 2 { + t.Fatalf("hermes post_tool_call groups = %d, want push + read", got) + } if !strings.Contains(string(data), "hermes session $s") { t.Fatal("hermes hook lacks its session-note label") } + if !strings.Contains(string(data), "bdrive read-log") { + t.Fatal("hermes read hook missing") + } // Idempotent. results, _ = Install(t.TempDir(), []string{"hermes"}) @@ -231,6 +276,47 @@ func TestHookCommandExtraction(t *testing.T) { } } +// The read hook must hand the event JSON through to `bdrive read-log` +// untouched — the binary does the parsing, the shell only guards. +func TestReadHookCommand(t *testing.T) { + if _, err := os.Stat("/bin/sh"); err != nil { + t.Skip("no /bin/sh") + } + dir := t.TempDir() + os.MkdirAll(filepath.Join(dir, ".bdrive"), 0o755) + bin := filepath.Join(dir, "bin") + os.MkdirAll(bin, 0o755) + fake := "#!/bin/sh\necho \"$@\" > \"" + dir + "/args.txt\"\ncat > \"" + dir + "/stdin.txt\"\n" + os.WriteFile(filepath.Join(bin, "bdrive"), []byte(fake), 0o755) + + payload := `{"session_id":"abc","tool_name":"Read","tool_input":{"file_path":"/x/wiki/a.md"}}` + sh := "cd " + dir + " && PATH=" + bin + ":$PATH " + readHookCommand() + if err := runShell(t, sh, payload); err != nil { + t.Fatal(err) + } + args, err := os.ReadFile(filepath.Join(dir, "args.txt")) + if err != nil { + t.Fatalf("read hook never called bdrive: %v", err) + } + if !strings.Contains(string(args), "read-log .") { + t.Fatalf("bdrive argv = %q, want read-log .", args) + } + stdin, _ := os.ReadFile(filepath.Join(dir, "stdin.txt")) + if string(stdin) != payload { + t.Fatalf("event JSON not passed through: %q", stdin) + } + + // Outside a bdrive project the hook exits before invoking anything. + os.Remove(filepath.Join(dir, "args.txt")) + os.RemoveAll(filepath.Join(dir, ".bdrive")) + if err := runShell(t, sh, payload); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(dir, "args.txt")); !os.IsNotExist(err) { + t.Fatal("hook invoked bdrive outside a project") + } +} + func runShell(t *testing.T, script, stdin string) error { t.Helper() cmd := exec.Command("/bin/sh", "-c", script) diff --git a/internal/remote/http.go b/internal/remote/http.go index ed80d2b..9504598 100644 --- a/internal/remote/http.go +++ b/internal/remote/http.go @@ -248,4 +248,28 @@ func (b *httpBackend) putViaServer(ctx context.Context, key string, r io.Reader, return nil } +// ReportReads sends the device's queued agent reads to the hub's read +// ledger, where they count as agent traffic (actor = this device). +func (b *httpBackend) ReportReads(ctx context.Context, reads []ReadEvent) error { + body, err := json.Marshal(map[string]any{"reads": reads}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + b.base+"/api/p/"+b.project+"/reads", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := b.do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return httpError(resp) + } + return nil +} + func (b *httpBackend) Close() error { return nil } diff --git a/internal/remote/remote.go b/internal/remote/remote.go index 4d68152..59d44d2 100644 --- a/internal/remote/remote.go +++ b/internal/remote/remote.go @@ -52,6 +52,20 @@ type Backend interface { Close() error } +// ReadEvent is one agent file read reported to the hub for its read heatmap. +type ReadEvent struct { + Path string `json:"path"` + Time time.Time `json:"time,omitzero"` +} + +// ReadReporter is the optional read-telemetry capability, in the PutSigner +// mold: backends that sync through a hub report the device's agent reads so +// the heat view can split human from agent traffic. Object-store backends +// simply don't implement it — there is no hub to tell. +type ReadReporter interface { + ReportReads(ctx context.Context, reads []ReadEvent) error +} + // Open creates a backend from a remote URL. func Open(ctx context.Context, raw string) (Backend, error) { u, err := url.Parse(raw) diff --git a/internal/store/reads.go b/internal/store/reads.go new file mode 100644 index 0000000..f62e3cd --- /dev/null +++ b/internal/store/reads.go @@ -0,0 +1,107 @@ +package store + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "time" +) + +// The read spool queues agent tool reads observed by `bdrive read-log` (the +// agent read hook) until a sync cycle drains it to the hub, where they count +// as agent traffic in the read heatmap. Hooks only append locally — no +// network on the hook path — and the flush is best-effort: offline, the +// spool just waits. + +// ReadEvent is one observed read of a synced file (mount-relative path). +type ReadEvent struct { + Path string `json:"path"` + Time time.Time `json:"time"` +} + +// readSpoolMax caps the spool: past it new events are dropped rather than +// letting an unreachable hub grow telemetry without bound. +const readSpoolMax = 1 << 20 + +// readReportMax bounds one drained batch to what the hub accepts per report. +const readReportMax = 4096 + +func (s *Store) readSpoolPath() string { return filepath.Join(s.dir, "reads.jsonl") } +func (s *Store) readFlushPath() string { return filepath.Join(s.dir, "reads-flushing.jsonl") } + +// LogRead appends one read event to the spool. Single-line O_APPEND writes +// keep concurrent hook invocations from interleaving. +func (s *Store) LogRead(rel string) error { + if fi, err := os.Stat(s.readSpoolPath()); err == nil && fi.Size() > readSpoolMax { + return nil // spool full: drop, never grow unbounded + } + line, err := json.Marshal(ReadEvent{Path: rel, Time: time.Now().UTC()}) + if err != nil { + return err + } + f, err := os.OpenFile(s.readSpoolPath(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return err + } + defer f.Close() + _, err = f.Write(append(line, '\n')) + return err +} + +// PendingReads returns the queued batch awaiting report, deduplicated by path +// (latest time wins). The spool is rotated aside first, so events logged +// after this call land in a fresh spool; the batch survives until +// ClearPendingReads — a failed report is simply retried next cycle. +func (s *Store) PendingReads() ([]ReadEvent, error) { + if _, err := os.Stat(s.readFlushPath()); os.IsNotExist(err) { + if err := os.Rename(s.readSpoolPath(), s.readFlushPath()); err != nil { + if os.IsNotExist(err) { + return nil, nil // nothing queued + } + return nil, err + } + } + data, err := os.ReadFile(s.readFlushPath()) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + latest := map[string]time.Time{} + var order []string + for _, line := range bytes.Split(data, []byte("\n")) { + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var e ReadEvent + if json.Unmarshal(line, &e) != nil || e.Path == "" { + continue // torn or corrupt line; drop it + } + if _, ok := latest[e.Path]; !ok { + order = append(order, e.Path) + } + if e.Time.After(latest[e.Path]) { + latest[e.Path] = e.Time + } + } + if len(order) > readReportMax { + order = order[len(order)-readReportMax:] + } + out := make([]ReadEvent, 0, len(order)) + for _, p := range order { + out = append(out, ReadEvent{Path: p, Time: latest[p]}) + } + return out, nil +} + +// ClearPendingReads drops the batch PendingReads returned, after a +// successful report. +func (s *Store) ClearPendingReads() error { + err := os.Remove(s.readFlushPath()) + if os.IsNotExist(err) { + return nil + } + return err +} diff --git a/internal/store/reads_test.go b/internal/store/reads_test.go new file mode 100644 index 0000000..ab52bea --- /dev/null +++ b/internal/store/reads_test.go @@ -0,0 +1,111 @@ +package store + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func openTestStore(t *testing.T) *Store { + t.Helper() + s, err := Open(filepath.Join(t.TempDir(), "volume")) + if err != nil { + t.Fatal(err) + } + return s +} + +func TestReadSpool(t *testing.T) { + s := openTestStore(t) + + // Nothing queued: no batch, no error. + if evs, err := s.PendingReads(); err != nil || len(evs) != 0 { + t.Fatalf("empty spool = %v, %v", evs, err) + } + + // Repeat reads of one path dedupe to its latest event. + for i := 0; i < 3; i++ { + if err := s.LogRead("wiki/a.md"); err != nil { + t.Fatal(err) + } + } + if err := s.LogRead("b.md"); err != nil { + t.Fatal(err) + } + evs, err := s.PendingReads() + if err != nil { + t.Fatal(err) + } + if len(evs) != 2 || evs[0].Path != "wiki/a.md" || evs[1].Path != "b.md" { + t.Fatalf("batch = %+v, want deduped a.md + b.md", evs) + } + if evs[0].Time.IsZero() { + t.Fatal("events must carry their read time") + } + + // The batch survives until cleared — a failed report just retries — and + // reads logged meanwhile land in a fresh spool behind it. + if err := s.LogRead("c.md"); err != nil { + t.Fatal(err) + } + again, err := s.PendingReads() + if err != nil { + t.Fatal(err) + } + if len(again) != 2 || again[0].Path != "wiki/a.md" { + t.Fatalf("retry batch = %+v, want the same uncleared batch", again) + } + if err := s.ClearPendingReads(); err != nil { + t.Fatal(err) + } + next, err := s.PendingReads() + if err != nil { + t.Fatal(err) + } + if len(next) != 1 || next[0].Path != "c.md" { + t.Fatalf("post-clear batch = %+v, want just c.md", next) + } + s.ClearPendingReads() + if evs, _ := s.PendingReads(); len(evs) != 0 { + t.Fatalf("drained spool still returned %+v", evs) + } +} + +func TestReadSpoolSurvivesCorruptLines(t *testing.T) { + s := openTestStore(t) + s.LogRead("good.md") + f, err := os.OpenFile(s.readSpoolPath(), os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + t.Fatal(err) + } + f.WriteString(`{"path": "torn`) // a torn write + f.Close() + s.LogRead("also-good.md") + evs, err := s.PendingReads() + if err != nil { + t.Fatal(err) + } + // The torn line joins the next event's line; both are dropped, but the + // batch itself survives. + if len(evs) == 0 || evs[0].Path != "good.md" { + t.Fatalf("batch = %+v, want good.md to survive the torn line", evs) + } +} + +func TestReadSpoolCap(t *testing.T) { + s := openTestStore(t) + long := strings.Repeat("d", 1024) + for i := 0; i < 1100; i++ { // ~1.1 MB of events + if err := s.LogRead(long + "/" + string(rune('a'+i%26)) + ".md"); err != nil { + t.Fatal(err) + } + } + fi, err := os.Stat(s.readSpoolPath()) + if err != nil { + t.Fatal(err) + } + if fi.Size() > readSpoolMax+4096 { + t.Fatalf("spool grew past its cap: %d bytes", fi.Size()) + } +} diff --git a/internal/syncer/ignore.go b/internal/syncer/ignore.go index b142605..8e9adec 100644 --- a/internal/syncer/ignore.go +++ b/internal/syncer/ignore.go @@ -53,6 +53,14 @@ type pattern struct { negate bool } +// LoadFilter builds the filter for a folder from its .bdriveignore (if any) +// plus the include list from the .bdrive settings file — the exact rules the +// sync cycle applies, for callers outside the cycle (e.g. `bdrive read-log` +// deciding whether an agent-read path is part of the project). +func LoadFilter(folder string, include []string) (*Filter, error) { + return loadFilter(folder, include) +} + // loadFilter builds the filter for a folder from its .bdriveignore (if any) // plus the include list from the .bdrive settings file. func loadFilter(folder string, include []string) (*Filter, error) { diff --git a/internal/syncer/reads_flow_test.go b/internal/syncer/reads_flow_test.go new file mode 100644 index 0000000..69122b9 --- /dev/null +++ b/internal/syncer/reads_flow_test.go @@ -0,0 +1,101 @@ +package syncer + +import ( + "context" + "fmt" + "sync" + "testing" + + "github.com/runbear-io/beardrive/internal/remote" +) + +// readReportingRemote wraps a backend with the hub's ReadReporter capability, +// standing in for the https:// backend in the multi-device harness. +type readReportingRemote struct { + remote.Backend + + mu sync.Mutex + fail bool + reports [][]remote.ReadEvent +} + +func (r *readReportingRemote) ReportReads(_ context.Context, reads []remote.ReadEvent) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.fail { + return fmt.Errorf("hub unreachable") + } + cp := make([]remote.ReadEvent, len(reads)) + copy(cp, reads) + r.reports = append(r.reports, cp) + return nil +} + +func (r *readReportingRemote) setFail(v bool) { + r.mu.Lock() + defer r.mu.Unlock() + r.fail = v +} + +func (r *readReportingRemote) all() [][]remote.ReadEvent { + r.mu.Lock() + defer r.mu.Unlock() + return r.reports +} + +// TestAgentReadReporting drives the read spool through real sync cycles: the +// queued reads flush to a reporting hub (deduped), survive an unreachable hub +// and retry, and never disturb the sync result itself. +func TestAgentReadReporting(t *testing.T) { + hub := &readReportingRemote{Backend: sharedRemote(t)} + a := newDevice(t, "deva", hub) + write(t, a.Folder, "wiki/a.md", "content") + + // The agent read a.md twice and b.md once before this cycle. + a.Store.LogRead("wiki/a.md") + a.Store.LogRead("wiki/a.md") + a.Store.LogRead("b.md") + res := cycle(t, a) + if !res.Pushed { + t.Fatal("cycle should have pushed") + } + reports := hub.all() + if len(reports) != 1 || len(reports[0]) != 2 { + t.Fatalf("reports = %+v, want one deduped batch of 2", reports) + } + if reports[0][0].Path != "wiki/a.md" || reports[0][1].Path != "b.md" { + t.Fatalf("batch = %+v", reports[0]) + } + // Drained: an idle cycle reports nothing. + cycle(t, a) + if len(hub.all()) != 1 { + t.Fatal("empty spool still produced a report") + } + + // Hub down: the cycle still succeeds and the batch stays queued. + hub.setFail(true) + a.Store.LogRead("wiki/a.md") + if res := cycle(t, a); res.Offline { + t.Fatal("a failed read report must not mark the cycle offline") + } + if len(hub.all()) != 1 { + t.Fatal("failed report should not have landed") + } + // Hub back: the next cycle retries the same batch. + hub.setFail(false) + cycle(t, a) + reports = hub.all() + if len(reports) != 2 || len(reports[1]) != 1 || reports[1][0].Path != "wiki/a.md" { + t.Fatalf("retry reports = %+v", reports) + } + + // A backend without the capability (plain object store) is untouched by + // queued reads: the cycle runs, the spool just keeps waiting. + b := newDevice(t, "devb", sharedRemote(t)) + write(t, b.Folder, "x.md", "x") + b.Store.LogRead("x.md") + cycle(t, b) + if evs, err := b.Store.PendingReads(); err != nil || len(evs) != 1 { + t.Fatalf("spool on a hubless device = %v, %v; want the read still queued", evs, err) + } +} diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index 7f53e8e..869a588 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -199,6 +199,21 @@ func (s *Session) Cycle(ctx context.Context) (*Result, error) { } } + // 6. Drain the agent read spool to the hub (read heatmap telemetry). + // Strictly best-effort: a failed report keeps the batch queued for the + // next cycle and never fails — or even marks offline — this one. + if rr, ok := s.Backend.(remote.ReadReporter); ok && !res.Offline { + if evs, err := s.Store.PendingReads(); err == nil && len(evs) > 0 { + reads := make([]remote.ReadEvent, len(evs)) + for i, e := range evs { + reads[i] = remote.ReadEvent{Path: e.Path, Time: e.Time} + } + if rr.ReportReads(ctx, reads) == nil { + s.Store.ClearPendingReads() + } + } + } + if err := s.Store.SaveCache(s.mountID(), cache); err != nil { return nil, err }