feat(sync): agent read reporting — read-log, spool, hub report, hook matchers

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/<id>/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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01P5cxPQdSGJnjXCYY9GeWXt
This commit is contained in:
Snow Lee
2026-07-11 14:49:02 -07:00
co-authored by Claude Fable 5
parent 28dc882c66
commit 2ce714ac4d
12 changed files with 758 additions and 28 deletions
+1
View File
@@ -36,6 +36,7 @@ everything keeps working offline; changes sync when the remote is reachable.`,
shareCmd(),
stopCmd(),
syncCmd(),
readLogCmd(),
hooksCmd(),
statusCmd(),
logCmd(),
+131
View File
@@ -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
}
+93
View File
@@ -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)
}
}
+64 -25
View File
@@ -13,7 +13,10 @@
//
// The hook syncs the project and stamps changes with "<agent> session <id>"
// (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.<Event> 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.<Event> 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 {
+89 -3
View File
@@ -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)
+24
View File
@@ -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 }
+14
View File
@@ -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)
+107
View File
@@ -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
}
+111
View File
@@ -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())
}
}
+8
View File
@@ -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) {
+101
View File
@@ -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)
}
}
+15
View File
@@ -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
}