feat(ci): PARTIAL P7-CACHE-21: node cache restore over carrier A only

free/ci/cmd/agent/cache_hook.go

@@ -84,6 +84,7 @@ func cacheBefore(ctx context.Context, l protocol.Lease, root session.JobRoot) (s
 			_ = base.Close()
 		}
 	}()
+	fetchRestores(ctx, copyStore, l)
 	seen := map[string]bool{}
 	for _, restore := range l.Cache.Restore {
 		sp, ok := specs.Get(restore.Spec)
@@ -110,8 +111,8 @@ func cacheBefore(ctx context.Context, l protocol.Lease, root session.JobRoot) (s
 			if !valid {
 				break
 			}
-			// JobHook.Before has no carrier receiver. Until that protocol hook is
-			// available a missing copy is best effort; never read coordinator paths.
+			// fetchRestores filled the read-through copies; a blob still missing
+			// is a cold miss (nil fetch), and coordinator paths are never read.
 			f, err := copyStore.Open(ctx, l.Cache.Namespace, b.SHA256, b.Size, nil)
 			if err != nil {
 				valid = false

free/ci/cmd/agent/cache_lifecycle.go

@@ -196,3 +196,20 @@ func validateCachePlan(plan *protocol.LeaseCache) error {
 	}
 	return nil
 }
+
+// fetchRestores takes the lease's restore blobs from the coordinator over the
+// session's reverse blob path (carrier A) into the read-through copies. Without a
+// feed (carrier B, or no restore) nothing is fetched: copies already held are still
+// used, and a missing one is a cold miss that binds nothing.
+func fetchRestores(ctx context.Context, copies *remote.ReadThrough, l protocol.Lease) {
+	feed, ok := session.FeedFrom(ctx)
+	if !ok {
+		return
+	}
+	defer feed.Close()
+	var blobs []protocol.CacheBlob
+	for _, r := range l.Cache.Restore {
+		blobs = append(blobs, r.Blobs...)
+	}
+	(&remote.NodeRestorer{Copies: copies, Feed: feed, Realm: l.Cache.Namespace}).Fetch(ctx, blobs)
+}

free/ci/internal/agent/session/control.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/nself-org/plugins/free/ci/internal/protocol"
+	"github.com/nself-org/plugins/free/ci/internal/transport"
 )
 
 func (s *Session) onMessage(m protocol.Message) (done bool, err error) {
@@ -22,9 +23,13 @@ func (s *Session) onMessage(m protocol.Message) (done bool, err error) {
 		return s.onDrain(m, *b)
 	case *protocol.Offset:
 		s.onOffset(m, *b)
+	case *protocol.Digest:
+		s.onInbound(m, transport.BlobFrame{Digest: b})
+	case *protocol.Chunk:
+		s.onInbound(m, transport.BlobFrame{Chunk: b})
 	}
-	// heartbeat, ack, upgrade, chunk, digest, resume: carried by later
-	// tickets (P7-NODE-24 upgrade, P7-CACHE-19 cache restore); the session ignores them.
+	// heartbeat, ack, upgrade, resume: carried by later tickets (P7-NODE-24 upgrade);
+	// the session ignores them.
 	return false, nil
 }
 

free/ci/internal/agent/session/inbound.go

@@ -0,0 +1,80 @@
+package session
+
+// Purpose: route the coordinator's digest and chunk frames to the job hook that
+// restores a cache from them (P7-CACHE-21, Epic CACHE D7): the reverse of an
+// upload. The hook finds the lease's feed on its context (FeedFrom).
+// Inputs:  `digest` and `chunk` frames fenced to a running lease and its epoch.
+// Outputs: a bounded channel per lease, read through transport.BlobFeed.
+// Constraints: only a lease whose plan restores blobs, on a carrier that sends
+// coordinator frames (RestoreCarrier), gets a feed; any other frame is dropped.
+// The Run goroutine blocks on a full channel only while the hook still reads it
+// (backpressure on the carrier, ended by Close, the lease context or shutdown);
+// a frame for an unknown lease or another epoch is ignored.
+
+import (
+	"context"
+	"sync"
+
+	"github.com/nself-org/plugins/free/ci/internal/protocol"
+	"github.com/nself-org/plugins/free/ci/internal/transport"
+)
+
+// inboundBuffer is how many unread frames a lease keeps (32 chunks is under 24 MiB of base64).
+const inboundBuffer = 32
+
+// RestoreCarrier is implemented by a Carrier that delivers coordinator digest and
+// chunk frames to the session (carrier A). A carrier without it gets no feed.
+type RestoreCarrier interface{ BlobRestore() bool }
+
+// BlobRestore says the stream carrier delivers every coordinator frame.
+func (*streamCarrier) BlobRestore() bool { return true }
+
+type feedKey struct{}
+
+// FeedFrom returns the reverse blob feed of the lease a Before hook runs for.
+func FeedFrom(ctx context.Context) (transport.BlobFeed, bool) {
+	f, ok := ctx.Value(feedKey{}).(*feed)
+	if !ok {
+		return nil, false
+	}
+	return f, true
+}
+
+type feed struct {
+	s      *Session
+	al     *activeLease
+	frames chan transport.BlobFrame
+	closed chan struct{}
+	once   sync.Once
+}
+
+func (s *Session) newFeed(al *activeLease) *feed {
+	rc, ok := s.cfg.Carrier.(RestoreCarrier)
+	if !ok || !rc.BlobRestore() || al.lease.Cache == nil || len(al.lease.Cache.Restore) == 0 {
+		return nil
+	}
+	return &feed{s: s, al: al, frames: make(chan transport.BlobFrame, inboundBuffer), closed: make(chan struct{})}
+}
+
+func (f *feed) Frames() <-chan transport.BlobFrame { return f.frames }
+func (f *feed) Offset(o protocol.Offset) error     { return f.s.send("offset", o, f.al) }
+func (f *feed) Close()                             { f.once.Do(func() { close(f.closed) }) }
+
+// onInbound hands a digest or chunk frame to the lease's feed.
+func (s *Session) onInbound(m protocol.Message, fr transport.BlobFrame) {
+	l, ok := s.active[m.LeaseID]
+	if !ok || m.Epoch == nil || *m.Epoch != l.lease.Epoch || l.feed == nil {
+		return
+	}
+	select {
+	case <-l.feed.closed:
+		return
+	default:
+	}
+	select {
+	case l.feed.frames <- fr:
+	case <-l.feed.closed:
+	case <-l.ctxDone:
+	case <-s.closing:
+	}
+}

free/ci/internal/agent/session/inbound_test.go

@@ -0,0 +1,100 @@
+package session
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/nself-org/plugins/free/ci/internal/protocol"
+)
+
+// restoreCar is a fake carrier that, like carrier A, delivers coordinator blob frames.
+type restoreCar struct{ *fakeCarrier }
+
+func (c restoreCar) fake() *fakeCarrier { return c.fakeCarrier }
+func (restoreCar) BlobRestore() bool    { return true }
+
+func restoreLease(id string, epoch uint64) protocol.Lease {
+	l := testLease(id, epoch)
+	l.Cache = &protocol.LeaseCache{Namespace: "ns", Access: "ro", Restore: []protocol.CacheRestore{{Spec: "go", Key: "k", Dir: "go"}}}
+	return l
+}
+
+// feedHook records what its Before hook sees of the reverse blob path.
+type feedHook struct {
+	got   chan []protocol.Digest
+	noOne chan bool
+}
+
+func newFeedHook() (*feedHook, JobHook) {
+	f := &feedHook{got: make(chan []protocol.Digest, 1), noOne: make(chan bool, 1)}
+	return f, JobHook{Name: "feed", Before: func(ctx context.Context, l protocol.Lease, _ JobRoot) (JobSetup, error) {
+		feed, ok := FeedFrom(ctx)
+		if !ok {
+			f.noOne <- true
+			return JobSetup{}, nil
+		}
+		defer feed.Close()
+		var seen []protocol.Digest
+		for len(seen) < 2 {
+			select {
+			case fr := <-feed.Frames():
+				if fr.Digest != nil {
+					seen = append(seen, *fr.Digest)
+					_ = feed.Offset(protocol.Offset{Stream: fr.Digest.Stream, SHA256: fr.Digest.SHA256, Offset: 0})
+				}
+			case <-time.After(5 * time.Second):
+				seen = append(seen, protocol.Digest{SHA256: "timeout"})
+			}
+		}
+		f.got <- seen
+		return JobSetup{}, nil
+	}}
+}
+
+func TestInboundBlobFramesReachTheHookFenced(t *testing.T) {
+	f, hook := newFeedHook()
+	car := restoreCar{newFakeCarrier()}
+	r := startRigWith(t, t.TempDir(), car, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook} })
+	r.handshake()
+	r.lease(restoreLease("l1", 5))
+	dig := func(sha string) protocol.Digest { return protocol.Digest{Stream: "cache", SHA256: sha, Size: 3} }
+	// none of these may reach the hook: another epoch, another lease
+	r.car.pushFenced("digest", dig("stale"), 4, "l1")
+	r.car.pushFenced("digest", dig("other"), 5, "l2")
+	r.car.pushFenced("digest", dig("one"), 5, "l1")
+	r.car.pushFenced("chunk", protocol.Chunk{Stream: "cache", Offset: 0, Data: "AAAA", Digest: "one", Final: true}, 5, "l1")
+	r.car.pushFenced("digest", dig("two"), 5, "l1")
+	select {
+	case seen := <-f.got:
+		if len(seen) != 2 || seen[0].SHA256 != "one" || seen[1].SHA256 != "two" {
+			t.Fatalf("hook saw %+v", seen)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("hook never saw the frames")
+	}
+	off := r.car.next(t, "offset").Body.(protocol.Offset)
+	if off.SHA256 != "one" {
+		t.Fatalf("offset answer %+v", off)
+	}
+	if m := r.car.next(t, "offset"); m.LeaseID != "l1" || m.Epoch == nil || *m.Epoch != 5 {
+		t.Fatalf("offset not fenced to the lease: %+v", m)
+	}
+}
+
+func TestInboundNoFeedWithoutCarrierSupportOrPlan(t *testing.T) {
+	f, hook := newFeedHook()
+	r := startRig(t, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook} }) // a carrier that cannot deliver blobs
+	r.handshake()
+	r.lease(restoreLease("l1", 1))
+	if !<-f.noOne {
+		t.Fatal("a feed was offered on a carrier without blob delivery")
+	}
+	f2, hook2 := newFeedHook()
+	r2 := startRigWith(t, t.TempDir(), restoreCar{newFakeCarrier()}, okRunner(`{}`), func(c *Config) { c.Hooks = []JobHook{hook2} })
+	r2.handshake()
+	r2.lease(testLease("l2", 1)) // no cache plan
+	if !<-f2.noOne {
+		t.Fatal("a feed was offered to a lease without restore blobs")
+	}
+}

free/ci/internal/agent/session/lease.go

@@ -37,6 +37,8 @@ type activeLease struct {
 	lines     atomic.Uint64 // safe
 	lw        *lineWriter
 	offsets   chan protocol.Offset // safe: the coordinator's offset answers (offsets.go)
+	feed      *feed                // set before the job goroutine starts: the reverse blob feed (inbound.go)
+	ctxDone   <-chan struct{}      // the lease context's Done channel
 }
 
 func (l *activeLease) cancel() { l.cancelled.Store(true); l.cancelCtx() }
@@ -98,6 +100,8 @@ func (s *Session) onLease(m protocol.Message, l protocol.Lease) {
 	// cancel, a drain deadline or the agent fence (P7-NODE-09).
 	ctx, cancel := context.WithCancel(s.runCtx)
 	al := &activeLease{lease: l, cancelCtx: cancel, offsets: make(chan protocol.Offset, offsetBuffer), secWake: make(chan struct{}, 1), secVals: map[string]string{}, expires: exp, secSeen: map[string]bool{}}
+	al.ctxDone = ctx.Done()
+	al.feed = s.newFeed(al)
 	if l.RunnerCredential != nil {
 		al.addMask(l.RunnerCredential.Value)
 	}
@@ -181,11 +185,15 @@ func (s *Session) runLease(ctx context.Context, al *activeLease, slot *slots.Slo
 
 	var setup JobSetup
 	skip := ""
+	beforeCtx := ctx
+	if al.feed != nil {
+		beforeCtx = context.WithValue(ctx, feedKey{}, al.feed)
+	}
 	for _, h := range s.jobHooks {
 		if h.Before == nil {
 			continue
 		}
-		got, err := h.Before(ctx, l, root)
+		got, err := h.Before(beforeCtx, l, root)
 		if err != nil && ctx.Err() != nil {
 			skip = "cancelled" // the hook stopped because the lease was cancelled
 			break
@@ -196,6 +204,9 @@ func (s *Session) runLease(ctx context.Context, al *activeLease, slot *slots.Slo
 		}
 		mergeSetup(&setup, got)
 	}
+	if al.feed != nil {
+		al.feed.Close() // the hooks are done: later frames are dropped
+	}
 	waitCtx, stopWait := context.WithTimeout(ctx, max(al.expires.Sub(s.cfg.Clock.Now()), 0))
 	for waitCtx.Err() == nil {
 		al.secMu.Lock()

free/ci/internal/cache/remote/coord.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/nself-org/plugins/free/ci/internal/authz"
 	"github.com/nself-org/plugins/free/ci/internal/blob/cas"
@@ -48,6 +49,8 @@ type Binding struct {
 	Fence, Ready             func(context.Context) error
 	// Unordered preserves the full ref grace for carriers without ordered delivery.
 	Unordered bool
+	// RestoreWait bounds each wait for the node's offset answer (default DefaultRestoreWait).
+	RestoreWait time.Duration
 }
 
 type Lease struct {

free/ci/internal/cache/remote/integration_agent_test.go

@@ -113,7 +113,7 @@ func (a *agentSide) addNode(id string) (model.Capability, tls.Certificate) {
 }
 
 // runAgent delivers the rig's lease to a node over the agent pull carrier and returns the driver's result.
-func (r *rig) runAgent(n *node) error {
+func (r *rig) runAgent(n nodeSide) error {
 	r.t.Helper()
 	side := r.agentSide()
 	capab, cert := side.addNode("n-1")