feat(sync): agents hear what teammates changed before they overwrite it (BEA-127) (#144)

A teammate's agent rewrites a file and yours never hears about it — the only
trace is a `.bdrive-conflict-*` nobody opens. Now the turn-start hook names
the paths that arrived since the last turn: "re-read before editing".

The record lives in a spool (`internal/store/inbound.go`), not in Cycle's
Result, because the daemon usually materializes a peer's change seconds
before the turn starts — so the hook's own cycle sees nothing. materialize
appends every path it writes or removes, `bdrive sync --hook` drains it after
its cycle and renders it under each mount's own prefix (stripping the session
subpath when the run is inside a mount, dropping paths outside it).

Advisory only: nothing blocks, nothing prompts, no per-Write remote call. The
spool is capped, 0600, in the volume dir, and best-effort everywhere — a
spool failure never fails a cycle or a turn.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/store/inbound.go

@@ -0,0 +1,111 @@
+package store
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// The inbound spool queues the paths a cycle materialized from peers, until
+// the agent hook (`bdrive sync --hook`) drains it into the turn's context —
+// "these changed since your last turn, re-read before editing". It is a spool
+// rather than a field on Result because the daemon usually materializes the
+// peer's change seconds before the turn starts, so the hook's own cycle
+// reports nothing: the record has to outlive the cycle that made it.
+
+// InboundEvent is one path a cycle wrote or removed on a peer's behalf
+// (mount-relative).
+type InboundEvent struct {
+	Path    string    `json:"path"`
+	Deleted bool      `json:"deleted,omitempty"`
+	Time    time.Time `json:"time"`
+}
+
+// inboundSpoolMax caps the spool: a machine with no agent hooks never drains,
+// so past the cap new events are dropped rather than growing without bound.
+const inboundSpoolMax = 1 << 20
+
+// inboundDrainMax bounds one drained batch; the hook renders far fewer.
+const inboundDrainMax = 4096
+
+func (s *Store) inboundSpoolPath() string { return filepath.Join(s.dir, "inbound.jsonl") }
+func (s *Store) inboundDrainPath() string { return filepath.Join(s.dir, "inbound-draining.jsonl") }
+
+// LogInbound appends one materialized path to the spool. Single-line O_APPEND
+// writes keep the daemon and a concurrent CLI cycle from interleaving.
+func (s *Store) LogInbound(rel string, deleted bool) error {
+	if fi, err := os.Stat(s.inboundSpoolPath()); err == nil && fi.Size() > inboundSpoolMax {
+		return nil // spool full: drop, never grow unbounded
+	}
+	line, err := json.Marshal(InboundEvent{Path: rel, Deleted: deleted, Time: time.Now().UTC()})
+	if err != nil {
+		return err
+	}
+	// 0600: the spool is a list of this project's file paths.
+	f, err := os.OpenFile(s.inboundSpoolPath(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	_, err = f.Write(append(line, '\n'))
+	return err
+}
+
+// DrainInbound returns the queued batch and clears it, deduplicated by path
+// (latest event wins, so a path written and then deleted reports as deleted).
+// The spool is rotated aside first, so events logged during the drain land in
+// a fresh spool — the drain runs outside the volume flock, and a daemon on
+// the same mount may be appending.
+//
+// One call, unlike PendingReads/ClearPendingReads: those are two steps
+// because a read report can fail over the network and must be retried, and
+// rendering a string onto stdout cannot.
+func (s *Store) DrainInbound() ([]InboundEvent, error) {
+	if _, err := os.Stat(s.inboundDrainPath()); os.IsNotExist(err) {
+		if err := os.Rename(s.inboundSpoolPath(), s.inboundDrainPath()); err != nil {
+			if os.IsNotExist(err) {
+				return nil, nil // nothing queued
+			}
+			return nil, err
+		}
+	}
+	data, err := os.ReadFile(s.inboundDrainPath())
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		// Unreadable: drop it rather than wedging every future drain behind
+		// it. Losing a turn's list is the cheaper failure.
+		os.Remove(s.inboundDrainPath())
+		return nil, err
+	}
+	defer os.Remove(s.inboundDrainPath())
+
+	latest := map[string]InboundEvent{}
+	var order []string
+	for _, line := range bytes.Split(data, []byte("\n")) {
+		if len(bytes.TrimSpace(line)) == 0 {
+			continue
+		}
+		var e InboundEvent
+		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 prev, ok := latest[e.Path]; !ok || !e.Time.Before(prev.Time) {
+			latest[e.Path] = e
+		}
+	}
+	if len(order) > inboundDrainMax {
+		order = order[len(order)-inboundDrainMax:]
+	}
+	out := make([]InboundEvent, 0, len(order))
+	for _, p := range order {
+		out = append(out, latest[p])
+	}
+	return out, nil
+}

internal/store/inbound_test.go

@@ -0,0 +1,118 @@
+package store
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestInboundSpool(t *testing.T) {
+	s := openTestStore(t)
+
+	// Nothing queued: no batch, no error.
+	if evs, err := s.DrainInbound(); err != nil || len(evs) != 0 {
+		t.Fatalf("empty spool = %v, %v", evs, err)
+	}
+
+	if err := s.LogInbound("wiki/a.md", false); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.LogInbound("b.md", false); err != nil {
+		t.Fatal(err)
+	}
+	// A path written and then removed reports as deleted: latest wins.
+	if err := s.LogInbound("b.md", true); err != nil {
+		t.Fatal(err)
+	}
+	evs, err := s.DrainInbound()
+	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 wiki/a.md + b.md", evs)
+	}
+	if evs[0].Deleted || !evs[1].Deleted {
+		t.Fatalf("batch = %+v, want b.md marked deleted", evs)
+	}
+	if evs[0].Time.IsZero() {
+		t.Fatal("events must carry their time")
+	}
+
+	// The drain clears: a second run with no activity in between reports
+	// nothing.
+	if again, err := s.DrainInbound(); err != nil || len(again) != 0 {
+		t.Fatalf("second drain = %+v, %v, want empty", again, err)
+	}
+}
+
+func TestInboundSpoolSurvivesCorruptLines(t *testing.T) {
+	s := openTestStore(t)
+	s.LogInbound("good.md", false)
+	f, err := os.OpenFile(s.inboundSpoolPath(), os.O_WRONLY|os.O_APPEND, 0o644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.WriteString(`{"path": "torn`) // a torn write
+	f.Close()
+	s.LogInbound("also-good.md", false)
+	evs, err := s.DrainInbound()
+	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 TestInboundSpoolCap(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.LogInbound(long+"/"+string(rune('a'+i%26))+".md", false); err != nil {
+			t.Fatal(err)
+		}
+	}
+	fi, err := os.Stat(s.inboundSpoolPath())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fi.Size() > inboundSpoolMax+4096 {
+		t.Fatalf("spool grew past its cap: %d bytes", fi.Size())
+	}
+}
+
+// The spool is a plain file in the volume dir at 0600 — never in the working
+// folder, never synced.
+func TestInboundSpoolPermissions(t *testing.T) {
+	s := openTestStore(t)
+	if err := s.LogInbound("a.md", false); err != nil {
+		t.Fatal(err)
+	}
+	fi, err := os.Stat(s.inboundSpoolPath())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fi.Mode().Perm() != 0o600 {
+		t.Fatalf("spool mode = %v, want 0600", fi.Mode().Perm())
+	}
+}
+
+// An unreadable spool must not wedge every later drain behind it.
+func TestInboundSpoolUnreadableRecovers(t *testing.T) {
+	s := openTestStore(t)
+	if err := os.MkdirAll(s.inboundSpoolPath(), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.DrainInbound(); err == nil {
+		t.Fatal("unreadable spool should report its error")
+	}
+	if err := s.LogInbound("a.md", false); err != nil {
+		t.Fatal(err)
+	}
+	evs, err := s.DrainInbound()
+	if err != nil || len(evs) != 1 || evs[0].Path != "a.md" {
+		t.Fatalf("drain after failure = %+v, %v, want a.md", evs, err)
+	}
+}