feat(ci): wire the dirty-tree offload into nself ci --where (P7-AUTO-13)

.github/wiki/guides/ci/Auto-Placement-and-Pools.md

@@ -71,7 +71,7 @@ A status of record is the commit status other people and merge rules rely on. It
 
 ## Dirty-tree offload
 
-To run uncommitted work on another node, `nself ci` takes a snapshot: a commit of the tracked changes plus untracked files that git does not ignore, whose parent is `HEAD`. The snapshot lives under a private ref in the `refs/nself-ci/offload` namespace and is deleted when the run finishes, fails or is cancelled. Your index, branches and working tree are not touched. A clean tree offloads `HEAD` unchanged.
+To run uncommitted work on another node, `nself ci` takes a snapshot: a commit of the tracked changes (staged or not, including files added with `git add` and not yet committed) plus untracked files that git does not ignore, whose parent is `HEAD`. The snapshot lives under a private ref in the `refs/nself-ci/offload` namespace and is deleted when the run finishes, fails or is cancelled. Your index, branches and working tree are not touched. A clean tree offloads `HEAD` unchanged. The snapshot is only what the node checks out: the run records the commit it started from, marked dirty, so `ci runs reproduce` never depends on a snapshot that no ref keeps. The snapshot is taken for `--where <node>`, and for a pool, `auto` or `all` when at least one node is registered. A run with `--where local` or `--where hosted`, or with only hosted providers registered, keeps `HEAD`: a hosted runner fetches the revision from the forge, where a snapshot never exists. For the same reason, a job that placement gives to a hosted runner checks out `HEAD` without your uncommitted changes, and the run prints `dirty changes not included: hosted runner`.
 
 Secrets are kept out by path and size, not by reading content. Untracked files that look like secrets or are too large are left out and named in the output. A tracked file of that kind refuses the offload. You can preview what would be included:
 

free/ci/cmd/v15_offload.go

@@ -0,0 +1,94 @@
+package main
+
+// Purpose: the dirty-tree offload of a dev run (P7-AUTO-13, ADR 0032 D15).
+// When `nself ci --where <node|pool|auto|all>` runs on a tree with uncommitted
+// work, nodes check out a snapshot commit (internal/offload) held under a
+// private ref, so the NODE-03 bundle transport carries the user's exact tree;
+// the run still records the commit it started from, dirty. This file
+// registers the run's OffloadHook slot.
+// Inputs:  the run root (Options.Root), the --where target the invocation
+//          resolved (whereChosen), git through the exec funnel.
+// Outputs: an OffloadSnapshot whose Release deletes the private ref; the names
+//          of untracked files left out go to stderr, never their content.
+// Constraints: only a target that can reach a registered node takes a
+//          snapshot: no --where with nothing remote registered, `local`,
+//          `hosted`, and pool/auto/all with no node keep the HEAD path
+//          unchanged (a hosted runner fetches the revision from the forge,
+//          where a snapshot never exists). A clean tree keeps HEAD. A
+//          tracked secret-shaped or oversized file refuses the run (E807). The
+//          snapshot run stays dirty: unbound evidence, no status of record.
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"github.com/nself-org/plugins/free/ci/internal/offload"
+	"github.com/nself-org/plugins/free/ci/internal/report"
+	cirun "github.com/nself-org/plugins/free/ci/internal/run"
+	"github.com/nself-org/plugins/free/ci/internal/where"
+)
+
+// Seams for tests.
+var (
+	offloadCapture offload.Capture = report.Capture
+	offloadErr     io.Writer       // nil: os.Stderr
+)
+
+func init() {
+	cirun.RegisterOffload(cirun.SnapshotOffload{Build: buildOffload, Excluded: reportExcluded})
+}
+
+// offloadWanted reports whether a run with this --where target may execute on
+// a registered node, the only kind of runner that fetches the revision by
+// bundle. A pool, auto or all target with no node registered (only hosted
+// providers) keeps HEAD.
+func offloadWanted(ctx context.Context, kind string) bool {
+	switch kind {
+	case where.KindNode:
+		return true
+	case where.KindPool, where.KindAuto, where.KindAll:
+		return len(whereKnown(ctx, true).Nodes) > 0
+	}
+	return false
+}
+
+// buildOffload snapshots the dirty tree at root, or returns nil to keep HEAD.
+func buildOffload(ctx context.Context, root string) (*cirun.OffloadSnapshot, error) {
+	if root == "" || !offloadWanted(ctx, whereChosen.Kind) {
+		return nil, nil
+	}
+	var nonce [8]byte
+	if _, err := rand.Read(nonce[:]); err != nil {
+		return nil, err
+	}
+	scratch, err := os.MkdirTemp("", "nself-ci-offload-")
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = os.RemoveAll(scratch) }() // the temporary index and path list
+	capture := offloadCapture
+	s, err := offload.Take(ctx, capture, root, scratch, hex.EncodeToString(nonce[:]))
+	if err != nil {
+		return nil, err
+	}
+	if s.Clean {
+		return nil, nil
+	}
+	return &cirun.OffloadSnapshot{SHA: s.SHA, Excluded: s.Excluded,
+		Release: func(ctx context.Context) error { return offload.Release(ctx, capture, root, s) }}, nil
+}
+
+// reportExcluded names the untracked files the snapshot left out.
+func reportExcluded(names []string) {
+	w := offloadErr
+	if w == nil {
+		w = os.Stderr
+	}
+	fmt.Fprintf(w, "nself ci: offload: not sent to the node (secret-shaped or over %d MiB, untracked): %s\n",
+		offload.MaxFileBytes>>20, strings.Join(names, ", "))
+}

free/ci/cmd/v15_offload_node_test.go

@@ -0,0 +1,247 @@
+package main
+
+// Purpose: a registered ssh node behind an in-process agent session, for the
+// cmd-level offload test (P7-AUTO-13). It is the test/sim foreground fixture
+// (P7-AUTO-28) reduced to what `nself ci --where <node>` needs: the node is in
+// the store under NSELF_CI_HOME, ssh.Default starts the session over pipes, and
+// the node stages the checkout bundle the real driver sends, as `serve --stdio`
+// does. No Docker, no network, no real ssh.
+// Inputs:  the run store directory and a body for the node's job.
+// Outputs: what the node's job found in the tree it was given.
+// Constraints: ssh.Default is restored after the test; tests do not run in parallel.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nself-org/cli/sdk/go/v2/remote"
+	"github.com/nself-org/plugins/free/ci/internal/agent/checkout"
+	"github.com/nself-org/plugins/free/ci/internal/agent/session"
+	"github.com/nself-org/plugins/free/ci/internal/agent/slots"
+	"github.com/nself-org/plugins/free/ci/internal/model"
+	"github.com/nself-org/plugins/free/ci/internal/nodes/registry"
+	"github.com/nself-org/plugins/free/ci/internal/planview"
+	"github.com/nself-org/plugins/free/ci/internal/protocol"
+	"github.com/nself-org/plugins/free/ci/internal/store"
+	"github.com/nself-org/plugins/free/ci/internal/transport/ssh"
+)
+
+type oflKey struct{}
+
+func (oflKey) KeyID() string        { return "k1" }
+func (oflKey) PublicString() string { return "pub" }
+
+// oflSeen is what the node's job found in its checkout.
+type oflSeen struct {
+	err   error
+	head  string
+	files map[string]string // name -> content for the probed names; a missing name is absent
+}
+
+// oflNode is one registered node and the agent behind it.
+type oflNode struct {
+	id, host, state string
+	probe           []string
+	seen            chan oflSeen
+}
+
+// oflStage is the node side of the checkout: the lease announcement and the
+// bundle's chunk frames go to the receiver and never reach the session.
+type oflStage struct {
+	session.Carrier
+	r *checkout.Receiver
+}
+
+func (c *oflStage) Recv() (protocol.Message, error) {
+	for {
+		m, err := c.Carrier.Recv()
+		if err != nil {
+			return m, err
+		}
+		switch b := m.Body.(type) {
+		case *protocol.Lease:
+			_ = c.r.Expect(*b)
+		case *protocol.Chunk:
+			if b.Stream != checkout.Stream {
+				return m, nil
+			}
+			epoch := uint64(0)
+			if m.Epoch != nil {
+				epoch = *m.Epoch
+			}
+			_ = c.r.Feed(m.LeaseID, epoch, *b)
+			continue
+		}
+		return m, nil
+	}
+}
+
+// serve runs one agent session; the job stages the bundle, records the tree and passes.
+func (n *oflNode) serve(ctx context.Context, table *slots.Table, in io.Reader, out io.Writer) error {
+	car, err := session.NewStreamCarrier(in, out, protocol.CarrierPolicy{Carrier: "A", Personal: true, OperatorOwned: true}, true)
+	if err != nil {
+		return err
+	}
+	recv := checkout.NewReceiver(filepath.Join(n.state, "checkout"), nil)
+	job := session.RunnerFunc(func(ctx context.Context, j session.JobRun) (session.JobResult, error) {
+		s := oflSeen{files: map[string]string{}}
+		src, err := recv.Stage(ctx, j.Lease, j.Root.Dir)
+		if err != nil {
+			s.err = err
+			n.seen <- s
+			return session.JobResult{}, err
+		}
+		if b, e := exec.Command("git", "-C", src, "rev-parse", "HEAD").Output(); e == nil {
+			s.head = strings.TrimSpace(string(b))
+		}
+		for _, name := range n.probe {
+			if b, e := os.ReadFile(filepath.Join(src, name)); e == nil {
+				s.files[name] = string(b)
+			}
+		}
+		n.seen <- s
+		j.Output([]byte("tree checked\n"))
+		ev, _ := json.Marshal(model.Evidence{Schema: "ci.evidence/v1", Revision: j.Lease.Checkout.Revision,
+			Attempt: &model.AttemptRef{ID: j.Lease.AttemptID, N: 1}, Result: "pass"})
+		return session.JobResult{Outcome: session.Outcome{State: "passed"}, Evidence: ev}, nil
+	})
+	s, err := session.New(session.Config{Carrier: &oflStage{Carrier: car, r: recv}, Key: oflKey{}, Slots: table, Runner: job, StateDir: n.state,
+		Hooks: []session.JobHook{}, Coordinator: session.Coordinator{ID: "coord", RootFingerprint: "fp"},
+		Facts: session.Facts{AgentVersion: minAgentVersion, MachineID: n.id}})
+	if err != nil {
+		return err
+	}
+	return s.Run(ctx)
+}
+
+// oflHosts maps a registered runner's ssh host to its in-process agent (one test at a time).
+var oflHosts map[string]*oflNode
+
+// oflStart is the ssh Starter: it picks the runner by destination host.
+func oflStart(ctx context.Context, tgt remote.Target, s string) (*ssh.Conn, error) {
+	for host, n := range oflHosts {
+		if strings.HasSuffix(tgt.Dest, "@"+host) {
+			return n.start(ctx, tgt, s)
+		}
+	}
+	return nil, fmt.Errorf("no node at %s", tgt.Dest)
+}
+
+// start opens a session on n: the "connection" is a pair of pipes.
+func (n *oflNode) start(_ context.Context, _ remote.Target, _ string) (*ssh.Conn, error) {
+	toAgentR, toAgentW := io.Pipe()
+	fromAgentR, fromAgentW := io.Pipe()
+	errR, errW := io.Pipe()
+	table, err := slots.Open(n.state, 2)
+	if err != nil {
+		return nil, err
+	}
+	actx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		e := n.serve(actx, table, toAgentR, fromAgentW)
+		_ = fromAgentW.Close()
+		_ = errW.Close()
+		done <- e
+	}()
+	kill := func() error { _ = fromAgentW.Close(); _ = toAgentR.Close(); cancel(); return nil }
+	return &ssh.Conn{Stdin: toAgentW, Stdout: fromAgentR, Stderr: errR, Wait: func() error { <-done; return nil }, Kill: kill}, nil
+}
+
+// oflRegisterNode registers node id in the store under home and points ssh.Default at it.
+func oflRegisterNode(t *testing.T, home, repo, id string, probe ...string) *oflNode {
+	t.Helper()
+	return oflRegisterRunner(t, home, repo, id, false, probe...)
+}
+
+// oflRegisterRunner is oflRegisterNode for an enrolled node, or, with hosted, for a
+// runner whose capability says it is hosted (sched.RunnerHosted): the dispatcher
+// then delivers its lease as a hosted one, over the same in-process session.
+func oflRegisterRunner(t *testing.T, home, repo, id string, hosted bool, probe ...string) *oflNode {
+	t.Helper()
+	n := &oflNode{id: id, host: id + ".runners.test", state: t.TempDir(), probe: probe, seen: make(chan oflSeen, 4)}
+	ctx := context.Background()
+	st, err := store.Open(filepath.Join(home, "state.db"), store.Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = st.Close() }()
+	pin := remote.PinnedHostKeys{Path: filepath.Join(home, "known_hosts")}
+	if err = pin.Add("nself-ci-"+id, n.host+" ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIDummyKeyForTestsOnlyDummyKeyForTestsOnly000"); err != nil {
+		t.Fatal(err)
+	}
+	c, err := planview.LocalCapability(planview.LocalFacts{Project: repo, Platform: runtime.GOOS + "/" + runtime.GOARCH, CPUs: 8, MemMB: 8192, Now: time.Now()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	lan, owned, uid := "lan", "owned", true
+	if hosted {
+		lan = "hosted"
+	}
+	c.Identity.ID, c.Identity.Name, c.Identity.Provider, c.Identity.Transport = id, id, "ssh", "ssh"
+	c.Identity.SSH = &model.CapabilitySSH{Hostname: n.host, Port: 22, User: "ci"}
+	c.SecretsEligible = true
+	c.Location.Kind.Value, c.Economics.Class.Value = &lan, &owned
+	c.Separation.UIDSeparation = model.Fact[bool]{Value: &uid, Source: "self", ObservedAt: time.Now(), Confidence: "known"}
+	doc, err := json.Marshal(c)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = st.NodePut(ctx, store.NodeRecord{ID: id, Name: id, State: "offline", Lifecycle: []byte("{}"), Trust: []byte("{}"),
+		DeployHost: []byte("{}"), Capability: doc, ObservedAt: time.Now().UnixNano()}, "test"); err != nil {
+		t.Fatal(err)
+	}
+	admin, err := store.NewAdminAuthority("test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	trust := model.CapabilityTrust{
+		Accepts:       oflAssigned([]model.TrustClass{"owner", "internal"}),
+		Isolation:     oflAssigned(model.Isolation("container")),
+		Network:       oflAssigned([]model.NetworkScope{"none", "internet", "restricted", "lan", "privileged"}),
+		SecretClasses: oflAssigned([]model.SecretClass{"none"}),
+		PrivacyZone:   oflAssigned(model.PrivacyZone("private-infrastructure")),
+	}
+	reg := registry.New(st)
+	if _, err = reg.SetTrust(ctx, id, trust, admin); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = reg.SetState(ctx, id, "online", "", "test"); err != nil {
+		t.Fatal(err)
+	}
+	if oflHosts == nil {
+		d := ssh.Default
+		home0, start0, ver0, hb0 := d.Home, d.Start, d.SSHVersion, d.Heartbeat
+		d.Home, d.Start, d.Heartbeat = home, oflStart, time.Second
+		d.SSHVersion = func(context.Context) (remote.Version, error) { return remote.Version{Major: 9, Minor: 6}, nil }
+		oflHosts = map[string]*oflNode{}
+		t.Cleanup(func() { d.Home, d.Start, d.SSHVersion, d.Heartbeat, oflHosts = home0, start0, ver0, hb0, nil })
+	}
+	oflHosts[n.host] = n
+	return n
+}
+
+func oflAssigned[T any](v T) model.Fact[T] {
+	return model.Fact[T]{Value: &v, Source: "assigned", ObservedAt: time.Now(), Confidence: "known"}
+}
+
+// wait returns what the node's job saw, or fails when no job reached the node.
+func (n *oflNode) wait(t *testing.T) oflSeen {
+	t.Helper()
+	select {
+	case s := <-n.seen:
+		return s
+	case <-time.After(60 * time.Second):
+		t.Fatal("no job reached the node")
+	}
+	panic("unreachable")
+}