feat(contribute): Executor seam for the contributor node + cloudy-agent

CONFORMANCE.md

@@ -37,6 +37,7 @@ Statuses are stated per the architecture's honesty rule. Built member-layer work
 - **Covenant and economy are stubs on main.** Their package docs carry the binding invariants; the built implementations are in the open, stacked PRs and must satisfy those docs to merge.
 - **Record domain-tag rename** (product naming) is pending and must land **before any durable persistence** — retagging after durable logs exist would be a rewrite.
 - **Named record residuals** (steganographic floor, timestamp covert channel, witness amnesia, liveness gap) are documented in the record package doc rather than pretended away.
+- **Contributor node — Executor seam built; runtime implementations pending.** `internal/contribute` fixes the node-contribution contract (Executor / ServiceExecutor / Registry / NodeContract) with two invariants enforced by tests: a scavenged node accepts only preemptible executors (owner-never-interrupted), and a Service workload requires pinned placement. `cmd/cloudy-agent` runs it end to end with a **real** Storage executor (opaque sealed shards + the `internal/storage` proof-of-possession) and **placeholder** Compute/Service executors. The heavy runtime port from the coordinator's transitional `internal/agent` — Docker executor, gopsutil hardware sampling, opt-out/allowlist/printers, and the agent→cloudyd→`/v0` relay — is the named next step and lands *behind* this seam without changing it.
 
 ## Dependency declaration
 

cmd/cloudy-agent/agent.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+
+	"github.com/NTARI-RAND/Cloudy/internal/contribute"
+)
+
+// ErrOwnerActive is returned when scavenged work is refused because the owner
+// has returned. It is a deferral, not a failure.
+var ErrOwnerActive = errors.New("cloudy-agent: owner active; scavenged work deferred")
+
+// Agent is the contributor daemon's control loop over the contribute seam. It
+// owns the owner-never-interrupted gate: on a scavenged node it stops
+// advertising and dispatching work while the owner is active. On a pinned node
+// the owner signal does not gate anything — the capacity is dedicated.
+//
+// This is the seam-side skeleton the migrated SoHoLINK agent plugs into: the
+// heartbeat/opt-out/telemetry loop calls Capabilities and Dispatch, and the
+// Docker/gopsutil/spiffe code lands as Executor implementations and a real
+// ownerActive probe, not as changes here.
+type Agent struct {
+	reg         *contribute.Registry
+	ownerActive func() bool
+	log         *slog.Logger
+}
+
+// NewAgent builds an agent. ownerActive reports whether the machine's owner is
+// currently using it (nil means "never active", e.g. a headless pinned box).
+func NewAgent(reg *contribute.Registry, ownerActive func() bool, log *slog.Logger) *Agent {
+	if ownerActive == nil {
+		ownerActive = func() bool { return false }
+	}
+	if log == nil {
+		log = slog.Default()
+	}
+	return &Agent{reg: reg, ownerActive: ownerActive, log: log}
+}
+
+// scavengedYield reports whether the owner gate is currently closing this node
+// to scavenged work.
+func (a *Agent) scavengedYield() bool {
+	return a.reg.Contract().Placement == contribute.Scavenged && a.ownerActive()
+}
+
+// Capabilities is what the agent advertises upstream this heartbeat. A
+// scavenged node advertises nothing while the owner is active — it has yielded
+// the machine entirely.
+func (a *Agent) Capabilities() []contribute.WorkloadKind {
+	if a.scavengedYield() {
+		return nil
+	}
+	return a.reg.Capabilities()
+}
+
+// Dispatch runs a one-shot job, refusing scavenged work while the owner is
+// active (ErrOwnerActive).
+func (a *Agent) Dispatch(ctx context.Context, job contribute.AssignedJob) (contribute.Result, error) {
+	if a.scavengedYield() {
+		return contribute.Result{JobID: job.ID, OK: false, Detail: "owner active"}, ErrOwnerActive
+	}
+	return a.reg.Dispatch(ctx, job)
+}
+
+// ServeService runs the node's pinned service (if any) until ctx is cancelled.
+// Services run on pinned capacity, so the owner gate does not apply.
+func (a *Agent) ServeService(ctx context.Context, job contribute.AssignedJob) error {
+	svc, ok := a.reg.Service()
+	if !ok {
+		return contribute.ErrNoExecutor
+	}
+	a.log.Info("service starting", "node", a.reg.Contract().NodeID, "job", job.ID)
+	err := svc.Serve(ctx, job)
+	a.log.Info("service stopped", "node", a.reg.Contract().NodeID, "job", job.ID, "err", err)
+	return err
+}

cmd/cloudy-agent/agent_test.go

@@ -0,0 +1,128 @@
+package main
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"github.com/NTARI-RAND/Cloudy/internal/contribute"
+)
+
+func pinnedAgent(t *testing.T, ownerActive func() bool) *Agent {
+	t.Helper()
+	reg, err := contribute.NewRegistry(contribute.NodeContract{
+		NodeID: "test", Placement: contribute.Pinned,
+		Kinds: []contribute.WorkloadKind{contribute.Compute, contribute.Storage, contribute.Service},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.Register(echoExecutor{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.Register(newStorageExecutor()); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.RegisterService(&presenceService{}); err != nil {
+		t.Fatal(err)
+	}
+	return NewAgent(reg, ownerActive, nil)
+}
+
+func TestStorageStoreThenAuditProvesPossession(t *testing.T) {
+	agent := pinnedAgent(t, nil)
+	shard := []byte("opaque sealed bytes the host can never read but can prove it holds")
+
+	store, _ := json.Marshal(storageOp{Op: "store", ID: "o1", Data: base64.StdEncoding.EncodeToString(shard)})
+	res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s1", Kind: contribute.Storage, Payload: store})
+	if err != nil || !res.OK {
+		t.Fatalf("store: res=%+v err=%v", res, err)
+	}
+
+	audit, _ := json.Marshal(storageOp{Op: "audit", ID: "o1"})
+	res, err = agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s2", Kind: contribute.Storage, Payload: audit})
+	if err != nil || !res.OK {
+		t.Fatalf("audit should prove possession: res=%+v err=%v", res, err)
+	}
+
+	// Auditing an unknown shard fails.
+	missing, _ := json.Marshal(storageOp{Op: "audit", ID: "nope"})
+	if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s3", Kind: contribute.Storage, Payload: missing}); err == nil {
+		t.Fatal("audit of missing shard should error")
+	}
+}
+
+func TestComputeEchoRoutes(t *testing.T) {
+	agent := pinnedAgent(t, nil)
+	res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("ping")})
+	if err != nil || !res.OK || string(res.Output) != "ping" {
+		t.Fatalf("compute echo: res=%+v err=%v", res, err)
+	}
+}
+
+func TestPinnedNodeIgnoresOwnerGate(t *testing.T) {
+	agent := pinnedAgent(t, func() bool { return true }) // owner "active"
+	// Pinned capacity is dedicated: owner activity does not reduce it.
+	if caps := agent.Capabilities(); len(caps) != 3 {
+		t.Fatalf("pinned node should advertise all 3 kinds regardless of owner, got %v", caps)
+	}
+	if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("x")}); err != nil {
+		t.Fatalf("pinned dispatch should not be gated: %v", err)
+	}
+}
+
+func TestScavengedNodeYieldsToOwner(t *testing.T) {
+	reg, err := contribute.NewRegistry(contribute.NodeContract{
+		NodeID: "laptop", Placement: contribute.Scavenged,
+		Kinds: []contribute.WorkloadKind{contribute.Compute, contribute.Storage},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.Register(echoExecutor{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.Register(newStorageExecutor()); err != nil {
+		t.Fatal(err)
+	}
+
+	active := true
+	agent := NewAgent(reg, func() bool { return active }, nil)
+
+	// Owner present: advertise nothing, refuse work.
+	if caps := agent.Capabilities(); caps != nil {
+		t.Fatalf("scavenged node with owner active should advertise nothing, got %v", caps)
+	}
+	if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("x")}); !errors.Is(err, ErrOwnerActive) {
+		t.Fatalf("want ErrOwnerActive, got %v", err)
+	}
+
+	// Owner leaves: capacity returns.
+	active = false
+	if caps := agent.Capabilities(); len(caps) != 2 {
+		t.Fatalf("scavenged node with owner away should advertise 2 kinds, got %v", caps)
+	}
+	if res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c2", Kind: contribute.Compute, Payload: []byte("go")}); err != nil || !res.OK {
+		t.Fatalf("dispatch after owner leaves: res=%+v err=%v", res, err)
+	}
+}
+
+func TestServiceLifecycle(t *testing.T) {
+	agent := pinnedAgent(t, nil)
+	svc, ok := agent.reg.Service()
+	if !ok {
+		t.Fatal("expected a registered service")
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- agent.ServeService(ctx, contribute.AssignedJob{ID: "svc", Kind: contribute.Service}) }()
+	cancel()
+	if err := <-done; !errors.Is(err, context.Canceled) {
+		t.Fatalf("service should stop cleanly on cancel, got %v", err)
+	}
+	if svc.Healthy() {
+		t.Error("service should report unhealthy after shutdown")
+	}
+}

cmd/cloudy-agent/executors.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"context"
+	"sync"
+
+	"github.com/NTARI-RAND/Cloudy/internal/contribute"
+)
+
+// echoExecutor is a minimal, dependency-free Compute executor: it returns its
+// payload. It stands in for the Docker compute executor (the heavy port that
+// satisfies this same seam next) so the Compute route is proven without
+// dragging the docker toolchain into the module yet. Preemptible.
+type echoExecutor struct{}
+
+func (echoExecutor) Kind() contribute.WorkloadKind { return contribute.Compute }
+func (echoExecutor) Preemptible() bool             { return true }
+func (echoExecutor) Run(_ context.Context, job contribute.AssignedJob) (contribute.Result, error) {
+	return contribute.Result{JobID: job.ID, OK: true, Detail: "echo", Output: job.Payload}, nil
+}
+
+// presenceService is a demo long-running Service: it reports healthy for its
+// lifetime and shuts down cleanly when the context is cancelled. It exists to
+// prove the ServiceExecutor half of the seam and the pinned-capacity contract;
+// a real hosted service (a site, a store, a game server) implements the same
+// interface.
+type presenceService struct {
+	mu      sync.Mutex
+	healthy bool
+}
+
+func (p *presenceService) Kind() contribute.WorkloadKind { return contribute.Service }
+
+func (p *presenceService) Healthy() bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.healthy
+}
+
+func (p *presenceService) Serve(ctx context.Context, _ contribute.AssignedJob) error {
+	p.mu.Lock()
+	p.healthy = true
+	p.mu.Unlock()
+	<-ctx.Done()
+	p.mu.Lock()
+	p.healthy = false
+	p.mu.Unlock()
+	return ctx.Err()
+}

cmd/cloudy-agent/main.go

@@ -0,0 +1,130 @@
+// Command cloudy-agent is the contributor daemon a Cloudy member runs on a
+// machine that contributes capacity. It is the home the SoHoLINK internal/agent
+// code moves into (executor, allowlist, opt-out, printers, hardware sampling,
+// heartbeat, telemetry) — but built around the contribute seam so the runtime
+// implementations (Docker, gopsutil, an Android storage provider, GPU) are
+// pluggable Executors, not the definition.
+//
+// This entrypoint is honest about what it is today: the seam and control loop
+// are real and tested; the registered executors are a real Storage node (over
+// internal/storage's proof-of-possession), a placeholder Compute echo standing
+// in for the Docker executor, and a demo presence Service. The Docker/gopsutil/
+// spiffe port and the agent->cloudyd->/v0 relay are the named next steps; they
+// land behind this seam without changing it.
+package main
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"flag"
+	"log/slog"
+	"os"
+	"os/signal"
+	"time"
+
+	"github.com/NTARI-RAND/Cloudy/internal/contribute"
+)
+
+func main() {
+	var (
+		nodeID   = flag.String("node", "cloudy-node-0", "node identifier")
+		pinned   = flag.Bool("pinned", true, "pinned (dedicated) capacity; false = scavenged (idle-first)")
+		withSvc  = flag.Bool("service", true, "offer the demo presence service (requires -pinned)")
+		selftest = flag.Bool("selftest", true, "run a store->audit->echo self-test and exit")
+	)
+	flag.Parse()
+
+	log := slog.New(slog.NewTextHandler(os.Stderr, nil))
+
+	placement := contribute.Scavenged
+	kinds := []contribute.WorkloadKind{contribute.Compute, contribute.Storage}
+	if *pinned {
+		placement = contribute.Pinned
+		if *withSvc {
+			kinds = append(kinds, contribute.Service)
+		}
+	}
+
+	reg, err := contribute.NewRegistry(contribute.NodeContract{NodeID: *nodeID, Placement: placement, Kinds: kinds})
+	if err != nil {
+		log.Error("contract invalid", "err", err)
+		os.Exit(1)
+	}
+	if err := reg.Register(echoExecutor{}); err != nil {
+		log.Error("register compute", "err", err)
+		os.Exit(1)
+	}
+	if err := reg.Register(newStorageExecutor()); err != nil {
+		log.Error("register storage", "err", err)
+		os.Exit(1)
+	}
+	if *pinned && *withSvc {
+		if err := reg.RegisterService(&presenceService{}); err != nil {
+			log.Error("register service", "err", err)
+			os.Exit(1)
+		}
+	}
+
+	// ownerActive: no real hardware probe yet (that is the gopsutil port);
+	// pinned nodes are never gated, so this stub is correct for -pinned.
+	agent := NewAgent(reg, func() bool { return false }, log)
+	log.Info("cloudy-agent up", "node", *nodeID, "placement", placement.String(), "capabilities", capsToStrings(agent.Capabilities()))
+
+	if *selftest {
+		runSelfTest(context.Background(), agent, log)
+		return
+	}
+
+	// Long-running mode: serve the pinned service until interrupted.
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer stop()
+	if _, ok := reg.Service(); ok {
+		if err := agent.ServeService(ctx, contribute.AssignedJob{ID: "presence-0", Kind: contribute.Service}); err != nil && ctx.Err() == nil {
+			log.Error("service failed", "err", err)
+			os.Exit(1)
+		}
+	} else {
+		<-ctx.Done()
+	}
+}
+
+func capsToStrings(ks []contribute.WorkloadKind) []string {
+	out := make([]string, len(ks))
+	for i, k := range ks {
+		out[i] = k.String()
+	}
+	return out
+}
+
+// runSelfTest exercises the seam end to end: store an opaque shard, prove
+// possession, and echo a compute job.
+func runSelfTest(ctx context.Context, agent *Agent, log *slog.Logger) {
+	shard := []byte("opaque-sealed-shard-bytes-the-node-cannot-read")
+	storePayload, _ := json.Marshal(storageOp{Op: "store", ID: "obj-1", Data: base64.StdEncoding.EncodeToString(shard)})
+	if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "s1", Kind: contribute.Storage, Payload: storePayload}); err != nil {
+		log.Error("store failed", "err", err)
+	} else {
+		log.Info("store", "ok", res.OK, "detail", res.Detail)
+	}
+
+	auditPayload, _ := json.Marshal(storageOp{Op: "audit", ID: "obj-1"})
+	if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "s2", Kind: contribute.Storage, Payload: auditPayload}); err != nil {
+		log.Error("audit failed", "err", err)
+	} else {
+		log.Info("audit", "ok", res.OK, "detail", res.Detail)
+	}
+
+	if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("hello")}); err != nil {
+		log.Error("compute failed", "err", err)
+	} else {
+		log.Info("compute", "ok", res.OK, "output", string(res.Output))
+	}
+
+	if svc, ok := agent.reg.Service(); ok {
+		sctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+		defer cancel()
+		_ = agent.ServeService(sctx, contribute.AssignedJob{ID: "svc-1", Kind: contribute.Service})
+		log.Info("service healthy after stop", "healthy", svc.Healthy())
+	}
+}