coulomb/fluid-core
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: coulomb:03ff7a8ad72f7f385d5ea9f13e4c68cc7582cbf7
Branches Tags
coulomb:main
..
compare: coulomb:d52dcc92a92d928fe801868a9e4379b9dda9101c
Branches Tags
coulomb:main

No commits in common. "03ff7a8ad72f7f385d5ea9f13e4c68cc7582cbf7" and "d52dcc92a92d928fe801868a9e4379b9dda9101c" have entirely different histories.
03ff7a8ad7 ... d52dcc92a9

17 changed files with 36 additions and 2609 deletions
Download patch file Download diff file Expand all files Collapse all files

160
cmd/fluid-control/main.go
View file

@ -1,160 +0,0 @@
// Command fluid-control serves the FLUID control-plane APIs.
//
// It publishes revisions and records governance decisions. It is not on the
// request path: ArchitectureBlueprint.md invariant 2 requires the data plane to
// keep serving when this process is down, and that separation is only real if
// nothing here is reachable from the gateway.
//
// This surface must never be exposed to interface consumers. It is the
// mechanism that evolves the interface in response to their behaviour, and a
// consumer able to reach it could drive its own adaptation.
package main
import (
"context"
"crypto/ed25519"
"encoding/base64"
"errors"
"flag"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
_ "modernc.org/sqlite"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/control"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
"github.com/tegwick/fluid-core/internal/publish"
"github.com/tegwick/fluid-core/internal/signing"
)
func main() {
if err := run(); err != nil {
log.Fatalf("fluid-control: %v", err)
}
}
func run() error {
var (
addr = flag.String("addr", "127.0.0.1:8081", "listen address")
store = flag.String("store", envOr("FLUID_STORE", "fluid.db"), "evidence store path")
iface = flag.String("interface", os.Getenv("FLUID_INTERFACE"), "interface identifier")
keyID = flag.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier")
keyFile = flag.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 private key file")
ephemeral = flag.Bool("ephemeral-key", false, "generate a throwaway signing key (development only)")
)
flag.Parse()
if *iface == "" {
return errors.New("--interface is required")
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
ev, err := evidence.OpenSQLite(ctx, *store)
if err != nil {
return err
}
defer ev.Close()
signer, err := loadSigner(*keyID, *keyFile, *ephemeral)
if err != nil {
return err
}
intents := intent.New(ev, contract.InterfaceID(*iface))
gate := policy.NewGate(policy.DefaultLimits())
pipeline, err := publish.New(publish.Options{
Gate: gate,
Signer: signer,
Store: ev,
Intents: intents,
})
if err != nil {
return err
}
srv := &http.Server{
Addr: *addr,
Handler: control.NewServer(control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate)).Routes(),
ReadHeaderTimeout: 10 * time.Second,
}
go func() {
<-ctx.Done()
shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_ = srv.Shutdown(shutdown)
}()
log.Printf("control plane for %s listening on %s (store %s, signing key %s)",
*iface, *addr, *store, signer.KeyID())
log.Printf("public key: %s", base64.StdEncoding.EncodeToString(signer.PublicKey()))
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
return err
}
return nil
}
// loadSigner resolves the signing key.
//
// An ephemeral key must be asked for explicitly. Generating one silently when
// none is configured would mean a deployment could sign revisions with a key
// nobody trusts and never notice until the router refused them.
func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) {
if keyFile != "" {
raw, err := os.ReadFile(keyFile)
if err != nil {
return nil, fmt.Errorf("read signing key: %w", err)
}
decoded, err := base64.StdEncoding.DecodeString(trimSpace(string(raw)))
if err != nil {
return nil, fmt.Errorf("signing key is not valid base64: %w", err)
}
return signing.NewSigner(keyID, ed25519.PrivateKey(decoded))
}
if ephemeral {
signer, _, err := signing.GenerateKey(keyID)
if err != nil {
return nil, err
}
log.Print("WARNING: using an ephemeral signing key; revisions signed now " +
"will not verify after a restart")
return signer, nil
}
return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development")
}
func trimSpace(s string) string {
start, end := 0, len(s)
for start < end && isSpace(s[start]) {
start++
}
for end > start && isSpace(s[end-1]) {
end--
}
return s[start:end]
}
func isSpace(b byte) bool {
return b == ' ' || b == '\t' || b == '\n' || b == '\r'
}
func envOr(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}

138
cmd/fluid/commands.go
View file

@ -2,8 +2,6 @@ package main
import (
"context"
"crypto/ed25519"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
@ -19,9 +17,6 @@ import (
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
"github.com/tegwick/fluid-core/internal/publish"
"github.com/tegwick/fluid-core/internal/signing"
)
// open connects to the evidence store for this invocation.
@ -132,13 +127,6 @@ func runRevision(ctx context.Context, g globals, args []string) error {
case "publish":
fs := newFlagSet("revision publish")
file := fs.String("file", "", "path to the revision descriptor (YAML or JSON)")
keyFile := fs.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 signing key file")
keyID := fs.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier")
ephemeral := fs.Bool("ephemeral-key", false, "sign with a throwaway key (development only)")
classes := fs.String("adaptation-classes", "", "comma-separated adaptation classes")
complexity := fs.Float64("complexity-delta", 0, "measured complexity impact")
share := fs.Float64("traffic-share", 0, "requested traffic share")
approvedBy := fs.String("approved-by", "", "authorizing operator; required by the default gate")
if err := fs.Parse(args[1:]); err != nil {
return err
}
@ -162,54 +150,43 @@ func runRevision(ctx context.Context, g globals, args []string) error {
if d.ID == "" {
return errors.New("descriptor has no revision id")
}
// The router refuses unsigned descriptors at runtime; refusing them here
// as well means an operator finds out at publish time rather than when
// traffic starts failing.
if d.Signature == nil {
fmt.Fprintln(os.Stderr,
"warning: descriptor is unsigned; the router will refuse it once signature enforcement is enabled")
}
// Publishing goes through the same pipeline the control plane uses.
// A CLI that could write a revision straight into the store would be a
// way around the deterministic policy gate, which would make the gate
// decorative for anyone with shell access.
signer, err := loadSigner(*keyID, *keyFile, *ephemeral)
body, err := json.Marshal(d)
if err != nil {
return err
}
intents := intent.New(store, contract.InterfaceID(iface))
pipeline, err := publish.New(publish.Options{
Gate: policy.NewGate(policy.DefaultLimits()),
Signer: signer,
Store: store,
Intents: intents,
})
if err != nil {
if err := store.PutRecord(ctx, contract.KindRevision, string(d.ID), body); err != nil {
return err
}
var approver *contract.Actor
if *approvedBy != "" {
approver = &contract.Actor{Type: contract.ActorTypeHuman, ID: *approvedBy}
}
candidate := publish.NewCandidate(d, contract.Actor{
Type: contract.ActorTypeHuman, ID: operator(),
})
verified, report, err := pipeline.Run(ctx, candidate, publish.PromotionRequest{
AdaptationClasses: parseClasses(*classes),
ComplexityDelta: *complexity,
RequestedTrafficShare: *share,
Approved: approver != nil,
ApprovedBy: approver,
})
if err != nil {
printReport(report)
return err
}
if err := pipeline.Publish(ctx, verified); err != nil {
if err := store.AppendEvent(ctx, contract.FluidEvent{
SchemaVersion: "0.1",
ID: contract.EventID(fmt.Sprintf("EV-pub-%s-%d", d.ID, time.Now().UnixNano())),
OccurredAt: time.Now().UTC(),
EntityType: contract.FluidEventEntityTypeRevision,
EntityID: string(d.ID),
EventType: "REVISION_PUBLISHED",
Actor: contract.Actor{Type: contract.ActorTypeHuman, ID: operator()},
Reason: fmt.Sprintf("published %s in state %s", d.ID, d.State),
}); err != nil {
return err
}
printReport(report)
fmt.Printf("\npublished %s (%s), governed by %s, signed by %s\n",
d.ID, d.State, d.Intent.Version, verified.Descriptor().Signature.KeyID)
// Binding the revision to its governing intent is what makes the later
// audit question answerable (Blueprint 27).
is := intent.New(store, contract.InterfaceID(iface))
if err := is.Bind(ctx, d.ID, d.Intent.Version); err != nil {
return fmt.Errorf("published, but binding to intent %s failed: %w", d.Intent.Version, err)
}
fmt.Printf("published %s (%s), governed by %s\n", d.ID, d.State, d.Intent.Version)
return nil
case "list":
@ -531,64 +508,3 @@ func operator() string {
}
return "unknown"
}
// parseClasses splits a comma-separated adaptation class list.
func parseClasses(s string) []contract.AdaptationClass {
if strings.TrimSpace(s) == "" {
return nil
}
var out []contract.AdaptationClass
for _, part := range strings.Split(s, ",") {
if p := strings.TrimSpace(part); p != "" {
out = append(out, contract.AdaptationClass(p))
}
}
return out
}
// printReport renders the pipeline outcome stage by stage, so a rejection says
// which gate refused and why rather than only that it failed.
func printReport(r publish.Report) {
if len(r.Stages) == 0 {
return
}
w := out()
fmt.Fprintln(w, "STAGE\tRESULT\tDETAIL")
for _, s := range r.Stages {
verdict := "pass"
if !s.Passed {
verdict = "FAIL"
}
fmt.Fprintf(w, "%s\t%s\t%s\n", s.Stage, verdict, oneLine(s.Detail))
}
_ = w.Flush()
}
// loadSigner resolves the signing key for a publish.
//
// An ephemeral key must be requested explicitly: signing with a key nobody
// trusts produces revisions the router will refuse, and finding that out at
// publish time is far better than at traffic time.
func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) {
if keyFile != "" {
raw, err := os.ReadFile(keyFile)
if err != nil {
return nil, fmt.Errorf("read signing key: %w", err)
}
decoded, err := base64.StdEncoding.DecodeString(strings.TrimSpace(string(raw)))
if err != nil {
return nil, fmt.Errorf("signing key is not valid base64: %w", err)
}
return signing.NewSigner(keyID, ed25519.PrivateKey(decoded))
}
if ephemeral {
signer, _, err := signing.GenerateKey(keyID)
if err != nil {
return nil, err
}
fmt.Fprintln(os.Stderr,
"warning: signing with an ephemeral key; this revision will not verify after the key is gone")
return signer, nil
}
return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development")
}

5
cmd/fluid/flags.go
View file

@ -20,7 +20,4 @@ func (f *flagSet) Bool(name string, value bool, usage string) *bool {
return f.fs.Bool(name, value, usage)
}
func (f *flagSet) Int(name string, value int, usage string) *int { return f.fs.Int(name, value, usage) }
func (f *flagSet) Float64(name string, value float64, usage string) *float64 {
return f.fs.Float64(name, value, usage)
}
func (f *flagSet) Parse(args []string) error { return f.fs.Parse(args) }
func (f *flagSet) Parse(args []string) error { return f.fs.Parse(args) }

95
internal/control/api.go
View file

@ -1,95 +0,0 @@
// Package control implements the FLUID control-plane APIs.
//
// ArchitectureBlueprint.md section 44 asks for a small set of internal control
// APIs. They are internal on purpose: this is the surface that publishes
// revisions and records governance decisions, and it must never be reachable by
// the consumers whose behaviour it evolves in response to.
//
// Nothing here is on the request path. The data plane keeps serving when this
// server is down (invariant 2), which is also why the CLI reads the evidence
// store directly rather than through these endpoints.
package control
import (
"encoding/json"
"errors"
"net/http"
"strings"
"github.com/tegwick/fluid-core/internal/evidence"
)
// Server exposes the control APIs over HTTP.
type Server struct {
revisions *RevisionAPI
intents *IntentAPI
}
// NewServer wires the control APIs.
func NewServer(rev *RevisionAPI, in *IntentAPI) *Server {
return &Server{revisions: rev, intents: in}
}
// Routes returns the control-plane mux.
func (s *Server) Routes() *http.ServeMux {
mux := http.NewServeMux()
mux.HandleFunc("/control/v1/revisions", s.revisions.handleCollection)
mux.HandleFunc("/control/v1/revisions/", s.revisions.handleItem)
mux.HandleFunc("/control/v1/intents", s.intents.handleCollection)
mux.HandleFunc("/control/v1/intents/", s.intents.handleItem)
mux.HandleFunc("/control/v1/intents/active", s.intents.handleActive)
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
})
return mux
}
// apiError is the control-plane error shape.
type apiError struct {
Error string `json:"error"`
Detail string `json:"detail,omitempty"`
Causes []string `json:"causes,omitempty"`
}
func writeJSON(w http.ResponseWriter, status int, body any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(body)
}
func writeError(w http.ResponseWriter, status int, msg string, causes ...string) {
writeJSON(w, status, apiError{Error: msg, Causes: causes})
}
// statusForStoreError maps store failures onto HTTP without leaking detail.
func statusForStoreError(err error) int {
if errors.Is(err, evidence.ErrNotFound) {
return http.StatusNotFound
}
return http.StatusInternalServerError
}
// pathTail returns the segment after prefix, or "" when there is none.
func pathTail(path, prefix string) string {
rest := strings.TrimPrefix(path, prefix)
rest = strings.Trim(rest, "/")
if rest == "" {
return ""
}
if i := strings.Index(rest, "/"); i >= 0 {
return rest[:i]
}
return rest
}
func decodeBody(r *http.Request, into any) error {
dec := json.NewDecoder(http.MaxBytesReader(nil, r.Body, 1<<20))
dec.DisallowUnknownFields()
if err := dec.Decode(into); err != nil {
return err
}
return nil
}

249
internal/control/api_test.go
View file

@ -1,249 +0,0 @@
package control
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
_ "modernc.org/sqlite"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
"github.com/tegwick/fluid-core/internal/publish"
"github.com/tegwick/fluid-core/internal/signing"
)
const intentDoc = `# Interface Evolution Intent
**Current operational authority mode:**
FLUID-2
`
func newServer(t *testing.T) (*http.ServeMux, *evidence.SQLStore) {
t.Helper()
ctx := context.Background()
store, err := evidence.OpenSQLite(ctx, filepath.Join(t.TempDir(), "e.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
intents := intent.New(store, "hall-publishing")
if _, err := intents.Put(ctx, "IEI-1", intentDoc); err != nil {
t.Fatal(err)
}
if err := intents.SetActive(ctx, "IEI-1"); err != nil {
t.Fatal(err)
}
signer, _, err := signing.GenerateKey("test-key")
if err != nil {
t.Fatal(err)
}
gate := policy.NewGate(policy.DefaultLimits())
pipeline, err := publish.New(publish.Options{
Gate: gate, Signer: signer, Store: store, Intents: intents,
})
if err != nil {
t.Fatal(err)
}
srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate))
return srv.Routes(), store
}
func descriptorJSON() contract.Revision {
pc := contract.RevisionPolicyPolicyCheckPassed
return contract.Revision{
SchemaVersion: "0.1",
ID: "R-2",
Interface: "hall-publishing",
State: contract.RevisionStateCandidate,
Contract: contract.RevisionContract{
Type: contract.RevisionContractTypeOpenapi,
Digest: contract.Digest("sha256:" + strings.Repeat("1", 64)),
},
Runtime: contract.RevisionRuntime{Upstream: "http://adapter:8080"},
Intent: contract.RevisionIntent{Version: "IEI-1"},
Policy: contract.RevisionPolicy{
Compatibility: contract.RevisionPolicyCompatibilityAdditive,
SecurityCheck: contract.RevisionPolicySecurityCheckPassed,
PolicyCheck: &pc,
},
}
}
func post(t *testing.T, mux *http.ServeMux, path string, body any) *httptest.ResponseRecorder {
t.Helper()
raw, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
mux.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, path, bytes.NewReader(raw)))
return rec
}
func TestCreateRevisionVerifiesAndPublishes(t *testing.T) {
mux, store := newServer(t)
rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{
Descriptor: descriptorJSON(),
Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"},
AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation},
ComplexityDelta: 0.2,
RequestedTrafficShare: 0.1,
Approved: true,
ApprovedBy: &contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"},
})
if rec.Code != http.StatusCreated {
t.Fatalf("status = %d, body %s", rec.Code, rec.Body.String())
}
var resp CreateRevisionResponse
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp.Descriptor == nil || resp.Descriptor.Signature == nil {
t.Fatal("published descriptor came back unsigned")
}
if !resp.Report.Passed() {
t.Errorf("report says not passed: %+v", resp.Report.Stages)
}
if _, err := store.Record(context.Background(), contract.KindRevision, "R-2"); err != nil {
t.Errorf("revision was not persisted: %v", err)
}
}
// TestRejectedCandidateIsAResultNotAFault: a rejection is normal (invariant 14)
// and must come back with its evidence rather than as a 500.
func TestRejectedCandidateIsAResultNotAFault(t *testing.T) {
mux, _ := newServer(t)
rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{
Descriptor: descriptorJSON(),
Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"},
AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation},
Approved: false, // the default gate requires approval
})
if rec.Code != http.StatusUnprocessableEntity {
t.Fatalf("status = %d, want 422; body %s", rec.Code, rec.Body.String())
}
var resp CreateRevisionResponse
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp.State != "REJECTED" {
t.Errorf("state = %q", resp.State)
}
failure, ok := resp.Report.FirstFailure()
if !ok {
t.Fatal("rejection carries no failing stage")
}
if failure.Stage != publish.StagePolicyCheck {
t.Errorf("failed at %s, want POLICY_CHECK", failure.Stage)
}
if len(failure.Evidence) == 0 {
t.Error("rejection carries no reasons")
}
}
func TestCreateRevisionValidatesInput(t *testing.T) {
mux, _ := newServer(t)
// A hypothesis id where a revision id belongs must not be accepted.
d := descriptorJSON()
d.ID = "H-2"
rec := post(t, mux, "/control/v1/revisions", CreateRevisionRequest{
Descriptor: d,
Origin: contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"},
})
if rec.Code != http.StatusBadRequest {
t.Errorf("mis-prefixed id: status = %d, want 400", rec.Code)
}
// Every candidate must name its origin: an artifact with no provenance
// cannot be audited later.
rec = post(t, mux, "/control/v1/revisions", CreateRevisionRequest{Descriptor: descriptorJSON()})
if rec.Code != http.StatusBadRequest {
t.Errorf("missing origin: status = %d, want 400", rec.Code)
}
}
func TestIntentEndpoints(t *testing.T) {
mux, _ := newServer(t)
rec := httptest.NewRecorder()
mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/control/v1/intents/active", nil))
if rec.Code != http.StatusOK {
t.Fatalf("active intent: status = %d, body %s", rec.Code, rec.Body.String())
}
var got IntentResponse
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
t.Fatal(err)
}
if got.Version != "IEI-1" || got.Mode != "FLUID-2" {
t.Errorf("active intent = %+v", got)
}
// Historical read by version.
rec = httptest.NewRecorder()
mux.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/control/v1/intents/IEI-1", nil))
if rec.Code != http.StatusOK {
t.Errorf("historical read: status = %d", rec.Code)
}
}
// TestRecordedIntentCannotBeRewritten: changing what a version says would
// change what already-published revisions were governed by.
func TestRecordedIntentCannotBeRewritten(t *testing.T) {
mux, _ := newServer(t)
rec := post(t, mux, "/control/v1/intents", RecordIntentRequest{
Version: "IEI-1",
Document: strings.Replace(intentDoc, "FLUID-2", "FLUID-5", 1),
})
if rec.Code != http.StatusConflict {
t.Errorf("status = %d, want 409; body %s", rec.Code, rec.Body.String())
}
}
func TestUnfilledTemplateIsRefused(t *testing.T) {
mux, _ := newServer(t)
rec := post(t, mux, "/control/v1/intents", RecordIntentRequest{
Version: "IEI-2",
Document: "mode is one of FLUID-0 FLUID-1 FLUID-2 FLUID-3 FLUID-4 FLUID-5 FLUID-6",
})
if rec.Code != http.StatusBadRequest {
t.Errorf("an unresolved template was accepted: status = %d", rec.Code)
}
}
func TestMethodsAreConstrained(t *testing.T) {
mux, _ := newServer(t)
for _, tc := range []struct{ method, path string }{
{http.MethodDelete, "/control/v1/revisions/R-2"},
{http.MethodPut, "/control/v1/intents/IEI-1"},
} {
rec := httptest.NewRecorder()
mux.ServeHTTP(rec, httptest.NewRequest(tc.method, tc.path, nil))
if rec.Code != http.StatusMethodNotAllowed {
t.Errorf("%s %s: status = %d, want 405", tc.method, tc.path, rec.Code)
}
}
}

128
internal/control/intent.go
View file

@ -1,128 +0,0 @@
package control
import (
"errors"
"net/http"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
)
// IntentAPI implements ArchitectureBlueprint.md section 44.5: read active
// intent, read historical intent, validate a candidate against intent, propose
// an amendment.
//
// Amendment is deliberately a proposal and not an edit. Blueprint section 22 is
// blunt about it: the Daimon must not silently expand its own mission, and
// intent changes require a separate governance process. This API records that
// an amendment was proposed; it never enacts one.
type IntentAPI struct {
store *intent.Store
gate *policy.Gate
}
// NewIntentAPI returns the intent API.
func NewIntentAPI(store *intent.Store, gate *policy.Gate) *IntentAPI {
return &IntentAPI{store: store, gate: gate}
}
// IntentResponse is a recorded intent version.
type IntentResponse struct {
Version string `json:"version"`
Digest contract.Digest `json:"digest"`
Mode string `json:"mode"`
Document string `json:"document"`
}
func (a *IntentAPI) handleCollection(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodPost:
a.record(w, r)
default:
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
}
}
func (a *IntentAPI) handleActive(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
return
}
v, err := a.store.Active(r.Context())
if err != nil {
if errors.Is(err, intent.ErrNoActive) {
writeError(w, http.StatusNotFound, "no active interface evolution intent")
return
}
writeError(w, http.StatusInternalServerError, "could not read active intent")
return
}
writeJSON(w, http.StatusOK, toResponse(v))
}
func (a *IntentAPI) handleItem(w http.ResponseWriter, r *http.Request) {
version := pathTail(r.URL.Path, "/control/v1/intents")
if version == "" || version == "active" {
a.handleActive(w, r)
return
}
switch r.Method {
case http.MethodGet:
v, err := a.store.Get(r.Context(), version)
if err != nil {
writeError(w, statusForStoreError(err), "intent version not found")
return
}
writeJSON(w, http.StatusOK, toResponse(v))
default:
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
}
}
// RecordIntentRequest records a new intent version.
type RecordIntentRequest struct {
Version string `json:"version"`
Document string `json:"document"`
Activate bool `json:"activate,omitempty"`
}
func (a *IntentAPI) record(w http.ResponseWriter, r *http.Request) {
var req RecordIntentRequest
if err := decodeBody(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "could not decode request", err.Error())
return
}
v, err := a.store.Put(r.Context(), req.Version, req.Document)
if err != nil {
if errors.Is(err, intent.ErrImmutable) {
// Rewriting a recorded version would change what already-published
// revisions were governed by, so it is a conflict, not a bad request.
writeError(w, http.StatusConflict, err.Error())
return
}
writeError(w, http.StatusBadRequest, err.Error())
return
}
if req.Activate {
if err := a.store.SetActive(r.Context(), v.Version); err != nil {
writeError(w, http.StatusInternalServerError, "recorded but not activated", err.Error())
return
}
}
writeJSON(w, http.StatusCreated, toResponse(v))
}
func toResponse(v intent.Version) IntentResponse {
return IntentResponse{
Version: v.Version,
Digest: v.Digest,
Mode: v.Mode.String(),
Document: v.Document,
}
}

176
internal/control/revision.go
View file

@ -1,176 +0,0 @@
package control
import (
"encoding/json"
"errors"
"net/http"
"sort"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/publish"
)
// RevisionAPI implements ArchitectureBlueprint.md section 44.1: create, verify,
// publish, set state, query lineage.
//
// Create and verify are one operation here rather than two. A revision that
// exists but has not been verified has no use and no authority, and offering it
// as a separate resource would invite callers to treat it as one.
type RevisionAPI struct {
store evidence.Store
pipeline *publish.Pipeline
}
// NewRevisionAPI returns the revision API.
func NewRevisionAPI(store evidence.Store, p *publish.Pipeline) *RevisionAPI {
return &RevisionAPI{store: store, pipeline: p}
}
// CreateRevisionRequest submits a candidate for verification and publication.
type CreateRevisionRequest struct {
Descriptor contract.Revision `json:"descriptor"`
Origin contract.Actor `json:"origin"`
AdaptationClasses []contract.AdaptationClass `json:"adaptation_classes,omitempty"`
ComplexityDelta float64 `json:"complexity_delta,omitempty"`
RequestedTrafficShare float64 `json:"requested_traffic_share,omitempty"`
Approved bool `json:"approved,omitempty"`
ApprovedBy *contract.Actor `json:"approved_by,omitempty"`
}
// CreateRevisionResponse reports the outcome, verified or not.
type CreateRevisionResponse struct {
Revision contract.RevisionID `json:"revision"`
State string `json:"state"`
Report publish.Report `json:"report"`
// Descriptor is returned only on success, with its signature attached.
Descriptor *contract.Revision `json:"descriptor,omitempty"`
}
func (a *RevisionAPI) handleCollection(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
a.list(w, r)
case http.MethodPost:
a.create(w, r)
default:
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
}
}
func (a *RevisionAPI) handleItem(w http.ResponseWriter, r *http.Request) {
id := pathTail(r.URL.Path, "/control/v1/revisions")
if id == "" {
writeError(w, http.StatusNotFound, "no revision named")
return
}
switch r.Method {
case http.MethodGet:
a.get(w, r, id)
default:
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
}
}
// create runs a candidate through the pipeline and publishes it if it passes.
func (a *RevisionAPI) create(w http.ResponseWriter, r *http.Request) {
var req CreateRevisionRequest
if err := decodeBody(r, &req); err != nil {
writeError(w, http.StatusBadRequest, "could not decode request", err.Error())
return
}
if req.Descriptor.ID == "" {
writeError(w, http.StatusBadRequest, "descriptor has no revision id")
return
}
if err := contract.RequireKind(string(req.Descriptor.ID), contract.KindRevision); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
if req.Origin.ID == "" {
writeError(w, http.StatusBadRequest, "every candidate must name its origin")
return
}
candidate := publish.NewCandidate(req.Descriptor, req.Origin)
verified, report, err := a.pipeline.Run(r.Context(), candidate, publish.PromotionRequest{
AdaptationClasses: req.AdaptationClasses,
ComplexityDelta: req.ComplexityDelta,
RequestedTrafficShare: req.RequestedTrafficShare,
Approved: req.Approved,
ApprovedBy: req.ApprovedBy,
})
if err != nil {
var rejected *publish.ErrRejected
if errors.As(err, &rejected) {
// A rejected candidate is a normal outcome (invariant 14), so it is
// reported as a result with its evidence rather than as a server
// fault. 422 says the request was well formed and the answer is no.
writeJSON(w, http.StatusUnprocessableEntity, CreateRevisionResponse{
Revision: candidate.ID(),
State: "REJECTED",
Report: report,
})
return
}
writeError(w, http.StatusInternalServerError, "pipeline failed", err.Error())
return
}
if err := a.pipeline.Publish(r.Context(), verified); err != nil {
writeError(w, http.StatusInternalServerError, "verified but not published", err.Error())
return
}
d := verified.Descriptor()
writeJSON(w, http.StatusCreated, CreateRevisionResponse{
Revision: d.ID,
State: string(d.State),
Report: report,
Descriptor: &d,
})
}
func (a *RevisionAPI) get(w http.ResponseWriter, r *http.Request, id string) {
body, err := a.store.Record(r.Context(), contract.KindRevision, id)
if err != nil {
writeError(w, statusForStoreError(err), "revision not found")
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(body)
}
// LineageEntry is one step in a revision's ancestry.
type LineageEntry struct {
Revision contract.RevisionID `json:"revision"`
State string `json:"state"`
Intent string `json:"intent"`
}
func (a *RevisionAPI) list(w http.ResponseWriter, r *http.Request) {
records, err := a.store.Records(r.Context(), contract.KindRevision)
if err != nil {
writeError(w, http.StatusInternalServerError, "could not list revisions")
return
}
out := make([]LineageEntry, 0, len(records))
for _, body := range records {
var d contract.Revision
if err := json.Unmarshal(body, &d); err != nil {
continue
}
out = append(out, LineageEntry{
Revision: d.ID,
State: string(d.State),
Intent: d.Intent.Version,
})
}
sort.Slice(out, func(i, j int) bool { return out[i].Revision < out[j].Revision })
writeJSON(w, http.StatusOK, map[string]any{"revisions": out})
}

224
internal/policy/gate.go
View file

@ -1,224 +0,0 @@
// Package policy implements the deterministic policy gate.
//
// ArchitectureBlueprint.md section 28.2: every candidate promotion must pass
// deterministic gates, and this gate is "the architectural boundary preventing
// agentic reasoning from becoming security policy".
//
// Nothing in this package may consult a model, call out to a service, or take
// an opinion as input. A gate that can be argued with is not a gate. The Daimon
// may reason about policy (section 48.7) but must not be the implementation of
// it, so every decision here is a pure function of the candidate, the governing
// intent, and explicitly configured limits.
package policy
import (
"fmt"
"sort"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/intent"
)
// Limits are the deterministic constraints a candidate must satisfy.
//
// They come from the interface evolution intent and from operator
// configuration, never from the candidate itself. A candidate that could raise
// its own ceiling would make the gate decorative.
type Limits struct {
// AllowedAdaptationClasses restricts what kinds of change may be promoted.
// Empty means every class is allowed.
AllowedAdaptationClasses []contract.AdaptationClass
// RequiredMode is the minimum authority the governing intent must declare
// for this promotion to be permitted at all.
RequiredMode intent.AuthorityMode
// MaxComplexityDelta caps the complexity a single candidate may add.
// Complexity is a budget (FluidAPIStandards.md section 22); a candidate may
// be rejected even when it increases local utility.
MaxComplexityDelta float64
// ComplexityLimitSet distinguishes "no limit" from "limit of zero".
ComplexityLimitSet bool
// MaxTrafficShare caps the exposure any single non-stable revision may take.
MaxTrafficShare float64
// RequireApproval demands a recorded human or policy authorization.
RequireApproval bool
// ProhibitedCompatibility lists compatibility classes that may never be
// promoted automatically, whatever else passes.
ProhibitedCompatibility []contract.RevisionPolicyCompatibility
}
// Decision is the gate's verdict.
type Decision struct {
Allowed bool `json:"allowed"`
Reasons []string `json:"reasons,omitempty"`
}
// Input is everything the gate is permitted to consider.
type Input struct {
Descriptor contract.Revision
// GoverningMode is the authority mode declared by the intent that governs
// this revision.
GoverningMode intent.AuthorityMode
// AdaptationClasses describes what the candidate changes.
AdaptationClasses []contract.AdaptationClass
// ComplexityDelta is the measured complexity impact.
ComplexityDelta float64
// RequestedTrafficShare is the exposure being asked for.
RequestedTrafficShare float64
// Approved reports whether a recorded authorization exists.
Approved bool
// ApprovedBy identifies the authorizing actor, when there is one.
ApprovedBy *contract.Actor
}
// Gate evaluates candidates against fixed limits.
type Gate struct{ limits Limits }
// NewGate returns a gate enforcing the given limits.
func NewGate(l Limits) *Gate { return &Gate{limits: l} }
// Limits returns the configured limits, for display and audit.
func (g *Gate) Limits() Limits { return g.limits }
// Evaluate applies every gate and returns a single decision.
//
// All checks run even after the first failure. A caller fixing one rejection
// only to hit the next is a worse experience than being told everything at
// once, and the full list is better evidence for the audit trail.
func (g *Gate) Evaluate(in Input) Decision {
var reasons []string
// Security status is checked first because it is the one condition where a
// pass by any other measure is irrelevant.
if in.Descriptor.Policy.SecurityCheck != contract.RevisionPolicySecurityCheckPassed {
reasons = append(reasons, fmt.Sprintf(
"security check is %q, must be %q",
in.Descriptor.Policy.SecurityCheck, contract.RevisionPolicySecurityCheckPassed))
}
if pc := in.Descriptor.Policy.PolicyCheck; pc != nil && *pc == contract.RevisionPolicyPolicyCheckFailed {
reasons = append(reasons, "policy check failed")
}
if !in.GoverningMode.Valid() {
reasons = append(reasons, "governing intent declares no valid authority mode")
} else if !in.GoverningMode.Allows(g.limits.RequiredMode) {
reasons = append(reasons, fmt.Sprintf(
"governing intent is at %s, but this promotion requires at least %s",
in.GoverningMode, g.limits.RequiredMode))
}
if len(g.limits.AllowedAdaptationClasses) > 0 {
allowed := map[contract.AdaptationClass]bool{}
for _, c := range g.limits.AllowedAdaptationClasses {
allowed[c] = true
}
for _, c := range in.AdaptationClasses {
if !allowed[c] {
reasons = append(reasons, fmt.Sprintf(
"adaptation class %q is not permitted here (permitted: %s)",
c, formatClasses(g.limits.AllowedAdaptationClasses)))
}
}
}
for _, prohibited := range g.limits.ProhibitedCompatibility {
if in.Descriptor.Policy.Compatibility == prohibited {
reasons = append(reasons, fmt.Sprintf(
"compatibility class %q may not be promoted under this policy", prohibited))
}
}
if g.limits.ComplexityLimitSet && in.ComplexityDelta > g.limits.MaxComplexityDelta {
reasons = append(reasons, fmt.Sprintf(
"complexity delta %.3f exceeds the budget of %.3f",
in.ComplexityDelta, g.limits.MaxComplexityDelta))
}
// Two ceilings apply to exposure, and the tighter one wins: the descriptor
// may declare its own maximum, and the gate imposes one. Taking the minimum
// means a descriptor can restrict itself further but never widen.
ceiling := g.limits.MaxTrafficShare
if r := in.Descriptor.Routing; r != nil && r.MaxTrafficShare != nil {
if declared := float64(*r.MaxTrafficShare); declared < ceiling || ceiling == 0 {
ceiling = declared
}
}
if ceiling > 0 && in.RequestedTrafficShare > ceiling {
reasons = append(reasons, fmt.Sprintf(
"requested traffic share %.2f exceeds the ceiling of %.2f",
in.RequestedTrafficShare, ceiling))
}
if g.limits.RequireApproval && !in.Approved {
reasons = append(reasons, "this promotion requires a recorded authorization and has none")
}
if in.Approved && in.ApprovedBy == nil {
reasons = append(reasons, "promotion is marked approved but names no authorizing actor")
}
// Generation authority is not promotion authority (Blueprint section 28.1).
// A candidate that a Daimon both produced and approved has had no
// independent check at all.
if in.Approved && in.ApprovedBy != nil && in.ApprovedBy.Type == contract.ActorTypeDaimon {
if !g.limits.RequiredMode.Allows(intent.ModeBoundedAutonomous) ||
!in.GoverningMode.Allows(intent.ModeBoundedAutonomous) {
reasons = append(reasons, fmt.Sprintf(
"authorization by a daimon requires at least %s authority, but the intent is at %s",
intent.ModeBoundedAutonomous, in.GoverningMode))
}
}
sort.Strings(reasons)
return Decision{Allowed: len(reasons) == 0, Reasons: reasons}
}
func formatClasses(cs []contract.AdaptationClass) string {
out := make([]string, len(cs))
for i, c := range cs {
out[i] = string(c)
}
sort.Strings(out)
return joinComma(out)
}
func joinComma(items []string) string {
switch len(items) {
case 0:
return "none"
case 1:
return items[0]
}
s := items[0]
for _, item := range items[1:] {
s += ", " + item
}
return s
}
// DefaultLimits returns a conservative starting configuration.
//
// The defaults refuse breaking changes, require approval, and permit only
// presentation and implementation adaptations — the two classes Blueprint
// section 37 identifies as safest. An interface that needs more should say so
// explicitly in its intent rather than inherit it.
func DefaultLimits() Limits {
return Limits{
AllowedAdaptationClasses: []contract.AdaptationClass{
contract.AdaptationClassPresentation,
contract.AdaptationClassImplementation,
},
RequiredMode: intent.ModeAdvisory,
MaxComplexityDelta: 1.0,
ComplexityLimitSet: true,
MaxTrafficShare: 0.25,
RequireApproval: true,
ProhibitedCompatibility: []contract.RevisionPolicyCompatibility{
contract.RevisionPolicyCompatibilityBreaking,
},
}
}

213
internal/policy/gate_test.go
View file

@ -1,213 +0,0 @@
package policy
import (
"strings"
"testing"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/intent"
)
func passing() contract.Revision {
pc := contract.RevisionPolicyPolicyCheckPassed
return contract.Revision{
ID: "R-2",
Interface: "hall-publishing",
State: contract.RevisionStateCandidate,
Policy: contract.RevisionPolicy{
Compatibility: contract.RevisionPolicyCompatibilityAdditive,
SecurityCheck: contract.RevisionPolicySecurityCheckPassed,
PolicyCheck: &pc,
},
}
}
func human() *contract.Actor {
return &contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}
}
func baseInput() Input {
return Input{
Descriptor: passing(),
GoverningMode: intent.ModeAdvisory,
AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation},
ComplexityDelta: 0.2,
RequestedTrafficShare: 0.10,
Approved: true,
ApprovedBy: human(),
}
}
func TestGateAllowsAConformingCandidate(t *testing.T) {
d := NewGate(DefaultLimits()).Evaluate(baseInput())
if !d.Allowed {
t.Fatalf("conforming candidate rejected: %v", d.Reasons)
}
}
func TestGateRefusesUnverifiedSecurity(t *testing.T) {
in := baseInput()
in.Descriptor.Policy.SecurityCheck = contract.RevisionPolicySecurityCheckPending
d := NewGate(DefaultLimits()).Evaluate(in)
if d.Allowed {
t.Fatal("a candidate with a pending security check was allowed")
}
if !mentions(d.Reasons, "security check") {
t.Errorf("reasons do not name the security check: %v", d.Reasons)
}
}
func TestGateRefusesBreakingChange(t *testing.T) {
in := baseInput()
in.Descriptor.Policy.Compatibility = contract.RevisionPolicyCompatibilityBreaking
if d := NewGate(DefaultLimits()).Evaluate(in); d.Allowed {
t.Fatal("a breaking change passed the default gate")
}
}
func TestGateEnforcesAuthorityMode(t *testing.T) {
limits := DefaultLimits()
limits.RequiredMode = intent.ModeExperimental
in := baseInput()
in.GoverningMode = intent.ModeAdvisory
d := NewGate(limits).Evaluate(in)
if d.Allowed {
t.Fatal("promotion requiring FLUID-4 was allowed under a FLUID-2 intent")
}
if !mentions(d.Reasons, "FLUID-2") {
t.Errorf("reasons do not name the governing mode: %v", d.Reasons)
}
in.GoverningMode = intent.ModeEvolutionary
if d := NewGate(limits).Evaluate(in); !d.Allowed {
t.Errorf("FLUID-6 should satisfy a FLUID-4 requirement: %v", d.Reasons)
}
}
func TestGateEnforcesComplexityBudget(t *testing.T) {
// FluidAPIStandards.md section 23: a candidate may be rejected even when it
// increases local utility.
in := baseInput()
in.ComplexityDelta = 9.0
d := NewGate(DefaultLimits()).Evaluate(in)
if d.Allowed {
t.Fatal("a candidate far over the complexity budget was allowed")
}
if !mentions(d.Reasons, "complexity") {
t.Errorf("reasons do not name complexity: %v", d.Reasons)
}
}
func TestGateTakesTheTighterTrafficCeiling(t *testing.T) {
limits := DefaultLimits()
limits.MaxTrafficShare = 0.50
in := baseInput()
share := contract.UnitInterval(0.10)
in.Descriptor.Routing = &contract.RevisionRouting{MaxTrafficShare: &share}
in.RequestedTrafficShare = 0.30
// The descriptor restricts itself below the gate's ceiling; the tighter of
// the two must win, or a descriptor's self-restriction would be advisory.
if d := NewGate(limits).Evaluate(in); d.Allowed {
t.Fatal("requested share exceeded the descriptor's own ceiling but was allowed")
}
// A descriptor must not be able to widen past the gate.
wide := contract.UnitInterval(0.99)
in.Descriptor.Routing = &contract.RevisionRouting{MaxTrafficShare: &wide}
in.RequestedTrafficShare = 0.80
if d := NewGate(limits).Evaluate(in); d.Allowed {
t.Fatal("a descriptor widened its own exposure past the gate ceiling")
}
}
func TestGateRequiresApproval(t *testing.T) {
in := baseInput()
in.Approved = false
in.ApprovedBy = nil
if d := NewGate(DefaultLimits()).Evaluate(in); d.Allowed {
t.Fatal("an unapproved candidate passed a gate that requires approval")
}
}
// TestDaimonCannotSelfApproveBelowBoundedAutonomy defends Blueprint 28.1:
// generation authority is not promotion authority. A candidate a Daimon both
// produced and approved has had no independent check.
func TestDaimonCannotSelfApproveBelowBoundedAutonomy(t *testing.T) {
in := baseInput()
in.ApprovedBy = &contract.Actor{Type: contract.ActorTypeDaimon, ID: "fluid-daimon/hall"}
d := NewGate(DefaultLimits()).Evaluate(in)
if d.Allowed {
t.Fatal("a daimon authorized its own promotion at FLUID-2")
}
if !mentions(d.Reasons, "daimon") {
t.Errorf("reasons do not name the daimon authorization: %v", d.Reasons)
}
// At bounded autonomy, with the policy demanding it, this is legitimate.
limits := DefaultLimits()
limits.RequiredMode = intent.ModeBoundedAutonomous
in.GoverningMode = intent.ModeBoundedAutonomous
if d := NewGate(limits).Evaluate(in); !d.Allowed {
t.Errorf("daimon authorization refused at FLUID-5: %v", d.Reasons)
}
}
func TestGateReportsEveryFailure(t *testing.T) {
// A caller fixing one rejection only to hit the next learns less than one
// told everything at once, and the audit trail wants the whole list.
in := baseInput()
in.Descriptor.Policy.SecurityCheck = contract.RevisionPolicySecurityCheckFailed
in.Descriptor.Policy.Compatibility = contract.RevisionPolicyCompatibilityBreaking
in.ComplexityDelta = 50
in.Approved = false
in.ApprovedBy = nil
in.AdaptationClasses = []contract.AdaptationClass{contract.AdaptationClassContract}
d := NewGate(DefaultLimits()).Evaluate(in)
if d.Allowed {
t.Fatal("a candidate failing five gates was allowed")
}
if len(d.Reasons) < 5 {
t.Errorf("expected at least five reasons, got %d: %v", len(d.Reasons), d.Reasons)
}
}
func TestGateIsDeterministic(t *testing.T) {
// The gate must be a pure function: same input, same verdict, same reasons
// in the same order. A gate whose output varies cannot be audited.
in := baseInput()
in.Descriptor.Policy.SecurityCheck = contract.RevisionPolicySecurityCheckFailed
in.AdaptationClasses = []contract.AdaptationClass{contract.AdaptationClassContract}
g := NewGate(DefaultLimits())
first := g.Evaluate(in)
for i := 0; i < 100; i++ {
again := g.Evaluate(in)
if again.Allowed != first.Allowed || len(again.Reasons) != len(first.Reasons) {
t.Fatalf("verdict varied between runs")
}
for j := range first.Reasons {
if again.Reasons[j] != first.Reasons[j] {
t.Fatalf("reason order varied: %q vs %q", first.Reasons[j], again.Reasons[j])
}
}
}
}
func mentions(reasons []string, substr string) bool {
for _, r := range reasons {
if strings.Contains(strings.ToLower(r), strings.ToLower(substr)) {
return true
}
}
return false
}

299
internal/publish/pipeline.go
View file

@ -1,299 +0,0 @@
package publish
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
"github.com/tegwick/fluid-core/internal/signing"
)
// Check runs one deterministic verification stage.
//
// Blueprint section 15 lists contract, regression, security, backend contract,
// performance, complexity and static policy checks. Implementations plug in
// here; the pipeline only cares that each returns a verdict and its evidence.
type Check interface {
// Stage names which pipeline step this check belongs to.
Stage() Stage
// Run reports whether the candidate passes, with evidence either way.
Run(context.Context, Candidate) StageResult
}
// CheckFunc adapts a function to Check.
type CheckFunc struct {
StageName Stage
Fn func(context.Context, Candidate) StageResult
}
// Stage implements Check.
func (c CheckFunc) Stage() Stage { return c.StageName }
// Run implements Check.
func (c CheckFunc) Run(ctx context.Context, cand Candidate) StageResult {
return c.Fn(ctx, cand)
}
// Pipeline implements the Blueprint section 35 publication pipeline.
//
// It is the only path from Candidate to Verified. Nothing else in the codebase
// constructs a Verified value, which is what makes "AI-generated artifacts are
// untrusted until verified" (invariant 9) a property of the code rather than a
// rule people are asked to remember.
type Pipeline struct {
checks []Check
gate *policy.Gate
signer *signing.Signer
store evidence.Store
intents *intent.Store
now func() time.Time
}
// Options configures a pipeline.
type Options struct {
// Checks run in pipeline order. A stage with no check is recorded as passed
// with an explicit note, so a gap in verification is visible in the report
// rather than invisible by omission.
Checks []Check
// Gate is the deterministic policy gate. Required.
Gate *policy.Gate
// Signer produces the signature that makes a descriptor routable. Required.
Signer *signing.Signer
// Store records the audit trail. Required.
Store evidence.Store
// Intents resolves the governing intent for a candidate. Required.
Intents *intent.Store
}
// New returns a pipeline.
func New(o Options) (*Pipeline, error) {
switch {
case o.Gate == nil:
return nil, fmt.Errorf("pipeline requires a policy gate")
case o.Signer == nil:
return nil, fmt.Errorf("pipeline requires a signer")
case o.Store == nil:
return nil, fmt.Errorf("pipeline requires an evidence store")
case o.Intents == nil:
return nil, fmt.Errorf("pipeline requires an intent store")
}
return &Pipeline{
checks: o.Checks,
gate: o.Gate,
signer: o.Signer,
store: o.Store,
intents: o.Intents,
now: time.Now,
}, nil
}
// PromotionRequest carries what the gate needs beyond the descriptor itself.
type PromotionRequest struct {
AdaptationClasses []contract.AdaptationClass
ComplexityDelta float64
RequestedTrafficShare float64
Approved bool
ApprovedBy *contract.Actor
}
// Run takes a candidate through every stage.
//
// On failure it returns the report alongside the error: a rejected candidate is
// normal (invariant 14), and the reasons are evidence worth keeping rather than
// an exception to discard.
func (p *Pipeline) Run(ctx context.Context, cand Candidate, req PromotionRequest) (Verified, Report, error) {
report := Report{Revision: cand.ID()}
byStage := map[Stage][]Check{}
for _, c := range p.checks {
byStage[c.Stage()] = append(byStage[c.Stage()], c)
}
governing, intentErr := p.intents.GoverningIntent(ctx, cand.ID())
if intentErr != nil {
// Fall back to the active intent: a candidate is normally bound to its
// intent at publish time, which has not happened yet.
governing, intentErr = p.intents.Active(ctx)
}
for _, stage := range Stages {
switch stage {
case StageSign:
// Signing is not a check; it is what the checks earn.
continue
case StagePublish:
continue
case StagePolicyCheck:
result := p.runPolicyGate(cand, req, governing, intentErr)
report.Stages = append(report.Stages, result)
p.record(ctx, cand, result)
if !result.Passed {
return Verified{}, report, &ErrRejected{Revision: cand.ID(), Failure: result}
}
continue
}
checks := byStage[stage]
if len(checks) == 0 {
// An unchecked stage is recorded as such. Silence here would let a
// pipeline with no security check look identical to one that passed.
result := StageResult{
Stage: stage,
Passed: true,
Detail: "no check configured for this stage",
}
report.Stages = append(report.Stages, result)
p.record(ctx, cand, result)
continue
}
for _, c := range checks {
result := c.Run(ctx, cand)
result.Stage = stage
report.Stages = append(report.Stages, result)
p.record(ctx, cand, result)
if !result.Passed {
return Verified{}, report, &ErrRejected{Revision: cand.ID(), Failure: result}
}
}
}
sig, err := p.signer.Sign(contract.RevisionDescriptorDocument{Revision: cand.Descriptor()})
if err != nil {
result := StageResult{Stage: StageSign, Passed: false, Detail: err.Error()}
report.Stages = append(report.Stages, result)
p.record(ctx, cand, result)
return Verified{}, report, &ErrRejected{Revision: cand.ID(), Failure: result}
}
signResult := StageResult{
Stage: StageSign,
Passed: true,
Detail: fmt.Sprintf("signed with key %s", sig.KeyID),
}
report.Stages = append(report.Stages, signResult)
p.record(ctx, cand, signResult)
return Verified{
descriptor: signedDescriptor(cand.Descriptor(), sig),
origin: cand.Origin(),
report: report,
}, report, nil
}
// runPolicyGate applies the deterministic gate.
func (p *Pipeline) runPolicyGate(cand Candidate, req PromotionRequest, governing intent.Version, intentErr error) StageResult {
if intentErr != nil {
// No governing intent means no constitutional basis for the change.
// Blueprint invariant 5 requires every revision to be governed by a
// specific intent version, so this is a refusal, not a warning.
return StageResult{
Stage: StagePolicyCheck,
Passed: false,
Detail: fmt.Sprintf("no governing interface evolution intent: %v", intentErr),
}
}
decision := p.gate.Evaluate(policy.Input{
Descriptor: cand.Descriptor(),
GoverningMode: governing.Mode,
AdaptationClasses: req.AdaptationClasses,
ComplexityDelta: req.ComplexityDelta,
RequestedTrafficShare: req.RequestedTrafficShare,
Approved: req.Approved,
ApprovedBy: req.ApprovedBy,
})
if decision.Allowed {
return StageResult{
Stage: StagePolicyCheck,
Passed: true,
Detail: fmt.Sprintf("passed under %s at %s", governing.Version, governing.Mode),
Evidence: []string{"intent:" + governing.Version},
}
}
return StageResult{
Stage: StagePolicyCheck,
Passed: false,
Detail: joinReasons(decision.Reasons),
Evidence: decision.Reasons,
}
}
// record appends a stage outcome to the audit trail.
//
// Every arrow in the pipeline should be independently observable (Blueprint
// section 46), so stages are recorded as they happen rather than summarized at
// the end — a pipeline that crashes mid-run still leaves evidence of how far it
// got.
func (p *Pipeline) record(ctx context.Context, cand Candidate, r StageResult) {
verdict := "PASSED"
if !r.Passed {
verdict = "FAILED"
}
_ = p.store.AppendEvent(ctx, contract.FluidEvent{
SchemaVersion: "0.1",
ID: contract.EventID(fmt.Sprintf("EV-%s-%s-%d",
cand.ID(), r.Stage, p.now().UnixNano())),
OccurredAt: p.now().UTC(),
EntityType: contract.FluidEventEntityTypeRevision,
EntityID: string(cand.ID()),
EventType: fmt.Sprintf("%s_%s", r.Stage, verdict),
Actor: cand.Origin(),
Reason: r.Detail,
})
}
// Publish records a verified revision and binds it to its governing intent.
//
// It takes a Verified rather than a Candidate: the type signature is the
// enforcement. There is no overload that accepts an unverified descriptor.
func (p *Pipeline) Publish(ctx context.Context, v Verified) error {
d := v.Descriptor()
if d.Signature == nil {
return ErrNotVerified
}
body, err := json.Marshal(d)
if err != nil {
return err
}
if err := p.store.PutRecord(ctx, contract.KindRevision, string(d.ID), body); err != nil {
return err
}
if err := p.intents.Bind(ctx, d.ID, d.Intent.Version); err != nil {
return fmt.Errorf("published %s but failed to bind its intent: %w", d.ID, err)
}
return p.store.AppendEvent(ctx, contract.FluidEvent{
SchemaVersion: "0.1",
ID: contract.EventID(fmt.Sprintf("EV-%s-PUBLISH-%d", d.ID, p.now().UnixNano())),
OccurredAt: p.now().UTC(),
EntityType: contract.FluidEventEntityTypeRevision,
EntityID: string(d.ID),
EventType: "REVISION_PUBLISHED",
Actor: v.Origin(),
Inputs: []string{d.Intent.Version},
Reason: fmt.Sprintf("published %s in state %s, governed by %s, signed by %s",
d.ID, d.State, d.Intent.Version, d.Signature.KeyID),
})
}
func joinReasons(reasons []string) string {
if len(reasons) == 0 {
return "rejected"
}
s := reasons[0]
for _, r := range reasons[1:] {
s += "; " + r
}
return s
}

283
internal/publish/pipeline_test.go
View file

@ -1,283 +0,0 @@
package publish
import (
"context"
"crypto/ed25519"
"errors"
"path/filepath"
"strings"
"testing"
_ "modernc.org/sqlite"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/evidence"
"github.com/tegwick/fluid-core/internal/intent"
"github.com/tegwick/fluid-core/internal/policy"
"github.com/tegwick/fluid-core/internal/signing"
)
const intentDoc = `# Interface Evolution Intent
**Current operational authority mode:**
FLUID-2
`
type fixture struct {
pipeline *Pipeline
store *evidence.SQLStore
intents *intent.Store
pub ed25519.PublicKey
keyID string
}
func newFixture(t *testing.T, limits policy.Limits, checks ...Check) *fixture {
t.Helper()
ctx := context.Background()
store, err := evidence.OpenSQLite(ctx, filepath.Join(t.TempDir(), "e.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
intents := intent.New(store, "hall-publishing")
if _, err := intents.Put(ctx, "IEI-1", intentDoc); err != nil {
t.Fatal(err)
}
if err := intents.SetActive(ctx, "IEI-1"); err != nil {
t.Fatal(err)
}
signer, pub, err := signing.GenerateKey("test-key")
if err != nil {
t.Fatal(err)
}
p, err := New(Options{
Checks: checks,
Gate: policy.NewGate(limits),
Signer: signer,
Store: store,
Intents: intents,
})
if err != nil {
t.Fatal(err)
}
return &fixture{pipeline: p, store: store, intents: intents, pub: pub, keyID: signer.KeyID()}
}
func candidate() Candidate {
pc := contract.RevisionPolicyPolicyCheckPassed
return NewCandidate(contract.Revision{
SchemaVersion: "0.1",
ID: "R-2",
Interface: "hall-publishing",
State: contract.RevisionStateCandidate,
Contract: contract.RevisionContract{
Type: contract.RevisionContractTypeOpenapi,
Digest: contract.Digest("sha256:" + strings.Repeat("1", 64)),
},
Runtime: contract.RevisionRuntime{Upstream: "http://adapter:8080"},
Intent: contract.RevisionIntent{Version: "IEI-1"},
Policy: contract.RevisionPolicy{
Compatibility: contract.RevisionPolicyCompatibilityAdditive,
SecurityCheck: contract.RevisionPolicySecurityCheckPassed,
PolicyCheck: &pc,
},
}, contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"})
}
func request() PromotionRequest {
return PromotionRequest{
AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation},
ComplexityDelta: 0.2,
RequestedTrafficShare: 0.10,
Approved: true,
ApprovedBy: &contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"},
}
}
func passingCheck(stage Stage) Check {
return CheckFunc{StageName: stage, Fn: func(context.Context, Candidate) StageResult {
return StageResult{Passed: true, Detail: "ok", Evidence: []string{"test:" + string(stage)}}
}}
}
func failingCheck(stage Stage, why string) Check {
return CheckFunc{StageName: stage, Fn: func(context.Context, Candidate) StageResult {
return StageResult{Passed: false, Detail: why}
}}
}
func TestPipelineVerifiesSignsAndPublishes(t *testing.T) {
ctx := context.Background()
f := newFixture(t, policy.DefaultLimits(),
passingCheck(StageContractCheck),
passingCheck(StageTest),
passingCheck(StageSecurityCheck),
)
verified, report, err := f.pipeline.Run(ctx, candidate(), request())
if err != nil {
t.Fatalf("pipeline rejected a conforming candidate: %v", err)
}
if !report.Passed() {
t.Fatalf("report says not passed: %+v", report.Stages)
}
// The signature is what the checks earn, and it must verify against the
// signer's key.
d := verified.Descriptor()
if d.Signature == nil {
t.Fatal("verified descriptor carries no signature")
}
v := signing.NewVerifier(map[string]ed25519.PublicKey{f.keyID: f.pub})
sig := &signing.Signature{
Algorithm: string(d.Signature.Algorithm),
KeyID: d.Signature.KeyID,
Value: d.Signature.Value,
}
if err := v.Verify(contract.RevisionDescriptorDocument{Revision: d}, sig); err != nil {
t.Fatalf("signature does not verify: %v", err)
}
if err := f.pipeline.Publish(ctx, verified); err != nil {
t.Fatalf("publish: %v", err)
}
// Publication must bind the revision to its governing intent, or the
// Blueprint 27 audit question becomes unanswerable.
got, err := f.intents.GoverningIntent(ctx, "R-2")
if err != nil {
t.Fatalf("no governing intent recorded: %v", err)
}
if got.Version != "IEI-1" {
t.Errorf("governing intent = %s, want IEI-1", got.Version)
}
}
func TestPipelineStopsAtFirstFailingStage(t *testing.T) {
ctx := context.Background()
f := newFixture(t, policy.DefaultLimits(),
passingCheck(StageContractCheck),
failingCheck(StageTest, "regression suite failed: 3 of 41"),
passingCheck(StageSecurityCheck),
)
_, report, err := f.pipeline.Run(ctx, candidate(), request())
if err == nil {
t.Fatal("pipeline accepted a candidate whose tests failed")
}
var rejected *ErrRejected
if !errors.As(err, &rejected) {
t.Fatalf("got %T, want *ErrRejected", err)
}
if rejected.Failure.Stage != StageTest {
t.Errorf("failed at %s, want TEST", rejected.Failure.Stage)
}
// Nothing after the failure should have run: a security check that never
// executed must not appear as passed.
for _, s := range report.Stages {
if s.Stage == StageSecurityCheck {
t.Error("a stage after the failure was executed")
}
}
}
// TestUncheckedStagesAreVisible guards against a pipeline with no security
// check looking identical to one that passed.
func TestUncheckedStagesAreVisible(t *testing.T) {
f := newFixture(t, policy.DefaultLimits())
_, report, err := f.pipeline.Run(context.Background(), candidate(), request())
if err != nil {
t.Fatal(err)
}
var security StageResult
for _, s := range report.Stages {
if s.Stage == StageSecurityCheck {
security = s
}
}
if security.Stage == "" {
t.Fatal("security stage missing from the report entirely")
}
if !strings.Contains(security.Detail, "no check configured") {
t.Errorf("an unchecked stage does not say so: %q", security.Detail)
}
}
// TestPolicyGateRejectionIsRecorded confirms the gate sits inside the pipeline
// rather than beside it.
func TestPolicyGateRejectionIsRecorded(t *testing.T) {
ctx := context.Background()
f := newFixture(t, policy.DefaultLimits())
req := request()
req.Approved = false
req.ApprovedBy = nil
_, _, err := f.pipeline.Run(ctx, candidate(), req)
if err == nil {
t.Fatal("an unapproved candidate was verified")
}
events, qerr := f.store.Events(ctx, evidence.EventFilter{EntityID: "R-2"})
if qerr != nil {
t.Fatal(qerr)
}
found := false
for _, ev := range events {
if ev.EventType == "POLICY_CHECK_FAILED" {
found = true
if !strings.Contains(ev.Reason, "authorization") {
t.Errorf("rejection reason not recorded usefully: %q", ev.Reason)
}
}
}
if !found {
t.Error("the policy gate rejection left no audit event")
}
}
// TestPublishRequiresVerification is the type-level check behind invariant 9.
// A zero Verified value has no signature and must be refused.
func TestPublishRequiresVerification(t *testing.T) {
f := newFixture(t, policy.DefaultLimits())
if err := f.pipeline.Publish(context.Background(), Verified{}); !errors.Is(err, ErrNotVerified) {
t.Errorf("publishing an unverified value returned %v, want ErrNotVerified", err)
}
}
func TestEveryStageLeavesAnEvent(t *testing.T) {
ctx := context.Background()
f := newFixture(t, policy.DefaultLimits(), passingCheck(StageTest))
if _, _, err := f.pipeline.Run(ctx, candidate(), request()); err != nil {
t.Fatal(err)
}
events, err := f.store.Events(ctx, evidence.EventFilter{EntityID: "R-2"})
if err != nil {
t.Fatal(err)
}
// Every arrow should be independently observable (Blueprint 46).
seen := map[string]bool{}
for _, ev := range events {
seen[ev.EventType] = true
}
for _, want := range []string{
"SOURCE_PASSED", "BUILD_PASSED", "CONTRACT_CHECK_PASSED",
"TEST_PASSED", "SECURITY_CHECK_PASSED", "POLICY_CHECK_PASSED", "SIGN_PASSED",
} {
if !seen[want] {
t.Errorf("no audit event for %s", want)
}
}
}

154
internal/publish/trust.go
View file

@ -1,154 +0,0 @@
// Package publish implements the revision publication pipeline and the trust
// transitions it enforces.
//
// ArchitectureBlueprint.md section 47 assigns different trust levels to
// different things: AI interpretation is advisory, generated code is an
// untrusted candidate, deterministic tests are verification evidence, and only
// a signed revision is a deployable artifact. Section 52 asks that this
// distinction stay visible in code and data models rather than living in
// reviewers' heads.
//
// The types here make it visible. A candidate cannot be published, because the
// publish path takes a Verified value, and the only way to obtain one is to
// pass verification. That is a weaker guarantee than a proof, but it means the
// unsafe path has to be written deliberately rather than reached by accident.
package publish
import (
"errors"
"fmt"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/signing"
)
// Candidate is an unverified revision descriptor.
//
// Whatever produced it — a human, a Builder, an LLM — it carries no authority.
// The field is unexported so that a Candidate can only be made through
// NewCandidate, and cannot be forged into a Verified value by struct literal
// from another package.
type Candidate struct {
descriptor contract.Revision
origin contract.Actor
}
// NewCandidate wraps a descriptor as an untrusted candidate.
func NewCandidate(d contract.Revision, origin contract.Actor) Candidate {
return Candidate{descriptor: d, origin: origin}
}
// Descriptor returns a copy of the candidate's descriptor for inspection.
//
// It is a copy on purpose: verification decides about a specific byte sequence,
// and handing out a mutable reference would let a caller change the artifact
// after it was judged.
func (c Candidate) Descriptor() contract.Revision { return c.descriptor }
// Origin reports who or what produced the candidate.
func (c Candidate) Origin() contract.Actor { return c.origin }
// ID reports the candidate's revision id.
func (c Candidate) ID() contract.RevisionID { return c.descriptor.ID }
// Verified is a candidate that has passed every deterministic gate and been
// signed. Only a Verified value may be published and routed.
type Verified struct {
descriptor contract.Revision
origin contract.Actor
report Report
}
// Descriptor returns the verified descriptor, signature included.
func (v Verified) Descriptor() contract.Revision { return v.descriptor }
// Origin reports who or what produced the underlying candidate.
func (v Verified) Origin() contract.Actor { return v.origin }
// Report returns the evidence that justified verification.
func (v Verified) Report() Report { return v.report }
// ID reports the revision id.
func (v Verified) ID() contract.RevisionID { return v.descriptor.ID }
// Stage names a step of the publication pipeline (Blueprint section 35).
type Stage string
const (
StageSource Stage = "SOURCE"
StageBuild Stage = "BUILD"
StageContractCheck Stage = "CONTRACT_CHECK"
StageTest Stage = "TEST"
StageSecurityCheck Stage = "SECURITY_CHECK"
StagePolicyCheck Stage = "POLICY_CHECK"
StageSign Stage = "SIGN"
StagePublish Stage = "PUBLISH"
)
// Stages is the pipeline in order.
var Stages = []Stage{
StageSource, StageBuild, StageContractCheck, StageTest,
StageSecurityCheck, StagePolicyCheck, StageSign, StagePublish,
}
// StageResult records what happened at one stage.
type StageResult struct {
Stage Stage `json:"stage"`
Passed bool `json:"passed"`
Detail string `json:"detail,omitempty"`
Evidence []string `json:"evidence,omitempty"`
}
// Report is the accumulated evidence for a publication attempt.
//
// It is retained whether or not the attempt succeeded. Failed candidates are
// normal (Blueprint invariant 14) and the record of why one was rejected is
// exactly the evidence a later hypothesis needs.
type Report struct {
Revision contract.RevisionID `json:"revision"`
Stages []StageResult `json:"stages"`
}
// Passed reports whether every recorded stage passed.
func (r Report) Passed() bool {
for _, s := range r.Stages {
if !s.Passed {
return false
}
}
return len(r.Stages) > 0
}
// FirstFailure returns the stage that stopped the pipeline.
func (r Report) FirstFailure() (StageResult, bool) {
for _, s := range r.Stages {
if !s.Passed {
return s, true
}
}
return StageResult{}, false
}
// ErrRejected reports a candidate that failed a gate.
type ErrRejected struct {
Revision contract.RevisionID
Failure StageResult
}
func (e *ErrRejected) Error() string {
return fmt.Sprintf("revision %s rejected at %s: %s", e.Revision, e.Failure.Stage, e.Failure.Detail)
}
// ErrNotVerified reports an attempt to publish something unverified.
var ErrNotVerified = errors.New("revision has not passed verification")
// signedDescriptor attaches a signature to a descriptor.
func signedDescriptor(d contract.Revision, sig signing.Signature) contract.Revision {
out := d
out.Signature = &contract.RevisionSignature{
Algorithm: contract.RevisionSignatureAlgorithm(sig.Algorithm),
KeyID: sig.KeyID,
Value: sig.Value,
}
return out
}

44
internal/runtime/registry.go
View file

@ -14,7 +14,6 @@ import (
"sync"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/signing"
)
// Registry is the gateway's cached view of published control-plane state.
@ -30,17 +29,9 @@ type Registry struct {
revisions map[contract.RevisionID]contract.Revision
policy contract.RoutingPolicy
hasPolicy bool
// verifier, when set, is applied to every descriptor before it is accepted.
verifier *signing.Verifier
}
// NewRegistry returns a registry that accepts unsigned descriptors.
//
// This is for development and tests. A deployment that routes real traffic
// should use NewVerifiedRegistry: ArchitectureBlueprint.md section 35 requires
// the router to accept only signed published descriptors, and a registry that
// takes anything makes the whole verification pipeline optional.
// NewRegistry returns an empty registry for one interface.
func NewRegistry(iface contract.InterfaceID) *Registry {
return &Registry{
iface: iface,
@ -48,14 +39,6 @@ func NewRegistry(iface contract.InterfaceID) *Registry {
}
}
// NewVerifiedRegistry returns a registry that refuses any descriptor which does
// not carry a signature from a trusted key.
func NewVerifiedRegistry(iface contract.InterfaceID, v *signing.Verifier) *Registry {
r := NewRegistry(iface)
r.verifier = v
return r
}
var (
// ErrUnknownRevision is returned for a revision the registry has never seen.
ErrUnknownRevision = errors.New("unknown revision")
@ -86,12 +69,6 @@ func (r *Registry) PutRevision(d contract.Revision) error {
return fmt.Errorf("revision %s: unknown state %q", d.ID, d.State)
}
if r.verifier != nil {
if err := r.verifyDescriptor(d); err != nil {
return fmt.Errorf("revision %s: %w", d.ID, err)
}
}
r.mu.Lock()
defer r.mu.Unlock()
r.revisions[d.ID] = d
@ -183,25 +160,6 @@ func (r *Registry) CheckRoutable(id contract.RevisionID, cohort contract.CohortI
return nil
}
// verifyDescriptor checks a descriptor's signature.
//
// The descriptor is verified in the exact form it was signed: the document
// wrapper included, the signature member excluded. Anything else would let a
// re-wrapped descriptor pass under a signature made for different bytes.
func (r *Registry) verifyDescriptor(d contract.Revision) error {
if d.Signature == nil {
return signing.ErrUnsigned
}
return r.verifier.Verify(
contract.RevisionDescriptorDocument{Revision: d},
&signing.Signature{
Algorithm: string(d.Signature.Algorithm),
KeyID: d.Signature.KeyID,
Value: d.Signature.Value,
},
)
}
func containsCohort(list []contract.CohortID, want contract.CohortID) bool {
for _, c := range list {
if c == want {

89
internal/runtime/registry_signing_test.go
View file

@ -1,89 +0,0 @@
package runtime
import (
"crypto/ed25519"
"errors"
"testing"
"github.com/tegwick/fluid-core/internal/contract"
"github.com/tegwick/fluid-core/internal/signing"
)
// TestVerifiedRegistryRefusesUnsigned is the router-side half of Blueprint 35.
// The pipeline signs; this is what makes the signature mean something.
func TestVerifiedRegistryRefusesUnsigned(t *testing.T) {
signer, pub, err := signing.GenerateKey("key-1")
if err != nil {
t.Fatal(err)
}
reg := NewVerifiedRegistry(testInterface, signing.NewVerifier(map[string]ed25519.PublicKey{"key-1": pub}))
unsigned := descriptor("R-1", contract.RevisionStateStable)
if err := reg.PutRevision(unsigned); !errors.Is(err, signing.ErrUnsigned) {
t.Fatalf("unsigned descriptor accepted: %v", err)
}
sig, err := signer.Sign(contract.RevisionDescriptorDocument{Revision: unsigned})
if err != nil {
t.Fatal(err)
}
signed := unsigned
signed.Signature = &contract.RevisionSignature{
Algorithm: contract.RevisionSignatureAlgorithm(sig.Algorithm),
KeyID: sig.KeyID,
Value: sig.Value,
}
if err := reg.PutRevision(signed); err != nil {
t.Fatalf("correctly signed descriptor refused: %v", err)
}
}
// TestVerifiedRegistryRefusesTamperedDescriptor covers the case that matters
// most: an attacker promoting a signed experiment to stable, or repointing it
// at their own adapter.
func TestVerifiedRegistryRefusesTamperedDescriptor(t *testing.T) {
signer, pub, _ := signing.GenerateKey("key-1")
reg := NewVerifiedRegistry(testInterface, signing.NewVerifier(map[string]ed25519.PublicKey{"key-1": pub}))
original := descriptor("R-1", contract.RevisionStateExperiment)
sig, _ := signer.Sign(contract.RevisionDescriptorDocument{Revision: original})
attach := func(d contract.Revision) contract.Revision {
d.Signature = &contract.RevisionSignature{
Algorithm: contract.RevisionSignatureAlgorithm(sig.Algorithm),
KeyID: sig.KeyID,
Value: sig.Value,
}
return d
}
promoted := original
promoted.State = contract.RevisionStateStable
if err := reg.PutRevision(attach(promoted)); !errors.Is(err, signing.ErrBadSignature) {
t.Errorf("a promoted descriptor was accepted: %v", err)
}
redirected := original
redirected.Runtime.Upstream = "http://attacker:8080"
if err := reg.PutRevision(attach(redirected)); !errors.Is(err, signing.ErrBadSignature) {
t.Errorf("a redirected descriptor was accepted: %v", err)
}
}
func TestVerifiedRegistryRefusesUntrustedKey(t *testing.T) {
rogue, _, _ := signing.GenerateKey("rogue")
_, trusted, _ := signing.GenerateKey("key-1")
reg := NewVerifiedRegistry(testInterface, signing.NewVerifier(map[string]ed25519.PublicKey{"key-1": trusted}))
d := descriptor("R-1", contract.RevisionStateStable)
sig, _ := rogue.Sign(contract.RevisionDescriptorDocument{Revision: d})
d.Signature = &contract.RevisionSignature{
Algorithm: contract.RevisionSignatureAlgorithm(sig.Algorithm),
KeyID: sig.KeyID,
Value: sig.Value,
}
if err := reg.PutRevision(d); !errors.Is(err, signing.ErrUnknownKey) {
t.Errorf("a descriptor signed by an untrusted key was accepted: %v", err)
}
}

210
internal/signing/signing.go
View file

@ -1,210 +0,0 @@
// Package signing implements descriptor and policy signatures.
//
// ArchitectureBlueprint.md section 35 requires the revision router to accept
// only signed or otherwise authenticated published descriptors. Section 28
// explains why: FLUID assumes generated or adaptive behaviour is untrusted
// until verified, and a signature is what carries the result of verification
// across a process boundary to the runtime that must act on it.
package signing
import (
"crypto/ed25519"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
)
// Algorithm is the only signature algorithm FLUID defines.
//
// One algorithm is deliberate. Negotiable algorithms invite downgrade, and a
// framework whose safety barrier can be talked down to a weaker primitive has
// no safety barrier.
const Algorithm = "ed25519"
var (
// ErrUnsigned reports a document carrying no signature.
ErrUnsigned = errors.New("document is not signed")
// ErrUnknownKey reports a signature from a key the verifier does not hold.
ErrUnknownKey = errors.New("signing key is not trusted")
// ErrBadSignature reports a signature that does not verify.
ErrBadSignature = errors.New("signature does not verify")
// ErrUnsupportedAlgorithm reports anything other than ed25519.
ErrUnsupportedAlgorithm = errors.New("unsupported signature algorithm")
)
// Signature is the detached signature attached to a signed document.
type Signature struct {
Algorithm string `json:"algorithm"`
KeyID string `json:"key_id"`
Value string `json:"value"`
SignedAt string `json:"signed_at,omitempty"`
}
// Canonicalize renders the signable form of a document.
//
// The signature member is removed before hashing, so a document signs its own
// content rather than its own signature. Map keys are sorted by Go's JSON
// encoder, which makes the byte sequence reproducible across processes and
// languages — a requirement, since an adapter or Daimon in another language
// must be able to produce a signature this verifier accepts.
func Canonicalize(document any) ([]byte, error) {
raw, err := json.Marshal(document)
if err != nil {
return nil, fmt.Errorf("canonicalize: %w", err)
}
var generic any
if err := json.Unmarshal(raw, &generic); err != nil {
return nil, fmt.Errorf("canonicalize: %w", err)
}
stripped := stripSignature(generic)
return json.Marshal(stripped)
}
// stripSignature removes every "signature" member, at any depth.
//
// Descriptors nest the payload under a "revision" key, and policies under
// "routing_policy", so the signature may sit one level down. Removing it
// wherever it appears keeps canonicalization independent of that shape.
func stripSignature(v any) any {
switch t := v.(type) {
case map[string]any:
out := make(map[string]any, len(t))
keys := make([]string, 0, len(t))
for k := range t {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
if k == "signature" {
continue
}
out[k] = stripSignature(t[k])
}
return out
case []any:
out := make([]any, len(t))
for i, item := range t {
out[i] = stripSignature(item)
}
return out
default:
return v
}
}
// Signer produces signatures.
//
// The signer holds a private key and nothing else. Blueprint section 28.1
// separates the builder identity from promotion rights: being able to sign an
// artifact is not the same authority as being able to publish one, and keeping
// this type free of any publishing capability is what makes that separable.
type Signer struct {
keyID string
key ed25519.PrivateKey
}
// NewSigner returns a signer for the given key.
func NewSigner(keyID string, key ed25519.PrivateKey) (*Signer, error) {
if strings.TrimSpace(keyID) == "" {
return nil, errors.New("signing key needs an id")
}
if len(key) != ed25519.PrivateKeySize {
return nil, fmt.Errorf("private key is %d bytes, want %d", len(key), ed25519.PrivateKeySize)
}
return &Signer{keyID: keyID, key: key}, nil
}
// KeyID reports which key this signer uses.
func (s *Signer) KeyID() string { return s.keyID }
// PublicKey returns the verifying half of the key pair.
func (s *Signer) PublicKey() ed25519.PublicKey {
return s.key.Public().(ed25519.PublicKey)
}
// Sign signs a document's canonical form.
func (s *Signer) Sign(document any) (Signature, error) {
payload, err := Canonicalize(document)
if err != nil {
return Signature{}, err
}
return Signature{
Algorithm: Algorithm,
KeyID: s.keyID,
Value: base64.StdEncoding.EncodeToString(ed25519.Sign(s.key, payload)),
}, nil
}
// Verifier checks signatures against a set of trusted keys.
type Verifier struct {
keys map[string]ed25519.PublicKey
}
// NewVerifier returns a verifier trusting the given keys, indexed by key id.
func NewVerifier(keys map[string]ed25519.PublicKey) *Verifier {
copied := make(map[string]ed25519.PublicKey, len(keys))
for id, k := range keys {
copied[id] = k
}
return &Verifier{keys: copied}
}
// Trust adds a key.
func (v *Verifier) Trust(keyID string, key ed25519.PublicKey) {
if v.keys == nil {
v.keys = map[string]ed25519.PublicKey{}
}
v.keys[keyID] = key
}
// Verify checks a document against its signature.
//
// An absent signature is refused rather than treated as "nothing to check".
// The distinction matters: a verifier that passes unsigned input is worse than
// no verifier, because it reports success.
func (v *Verifier) Verify(document any, sig *Signature) error {
if sig == nil {
return ErrUnsigned
}
if sig.Algorithm != Algorithm {
return fmt.Errorf("%w: %q", ErrUnsupportedAlgorithm, sig.Algorithm)
}
key, ok := v.keys[sig.KeyID]
if !ok {
return fmt.Errorf("%w: %q", ErrUnknownKey, sig.KeyID)
}
raw, err := base64.StdEncoding.DecodeString(sig.Value)
if err != nil {
return fmt.Errorf("%w: signature is not valid base64", ErrBadSignature)
}
payload, err := Canonicalize(document)
if err != nil {
return err
}
if !ed25519.Verify(key, payload, raw) {
return ErrBadSignature
}
return nil
}
// GenerateKey produces a new signing key pair. Intended for development and
// tests; production keys should come from the deployment's key management.
func GenerateKey(keyID string) (*Signer, ed25519.PublicKey, error) {
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
return nil, nil, err
}
s, err := NewSigner(keyID, priv)
if err != nil {
return nil, nil, err
}
return s, pub, nil
}

164
internal/signing/signing_test.go
View file

@ -1,164 +0,0 @@
package signing
import (
"crypto/ed25519"
"errors"
"testing"
)
type doc struct {
Revision struct {
ID string `json:"id"`
State string `json:"state"`
Upstream string `json:"upstream"`
Signature *Signature `json:"signature,omitempty"`
} `json:"revision"`
}
func newDoc(id, state string) doc {
var d doc
d.Revision.ID = id
d.Revision.State = state
d.Revision.Upstream = "http://adapter:8080"
return d
}
func TestSignAndVerify(t *testing.T) {
signer, pub, err := GenerateKey("key-1")
if err != nil {
t.Fatal(err)
}
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": pub})
d := newDoc("R-1", "stable")
sig, err := signer.Sign(d)
if err != nil {
t.Fatal(err)
}
if err := v.Verify(d, &sig); err != nil {
t.Fatalf("freshly signed document does not verify: %v", err)
}
}
// TestSignatureCoversContent is the property the whole barrier depends on: a
// tampered descriptor must not verify.
func TestSignatureCoversContent(t *testing.T) {
signer, pub, _ := GenerateKey("key-1")
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": pub})
d := newDoc("R-1", "experiment")
sig, _ := signer.Sign(d)
// Promoting a signed experiment to stable by editing the descriptor is
// exactly the attack the signature exists to stop.
tampered := d
tampered.Revision.State = "stable"
if err := v.Verify(tampered, &sig); !errors.Is(err, ErrBadSignature) {
t.Errorf("a tampered descriptor verified: %v", err)
}
redirected := d
redirected.Revision.Upstream = "http://attacker:8080"
if err := v.Verify(redirected, &sig); !errors.Is(err, ErrBadSignature) {
t.Errorf("a redirected upstream verified: %v", err)
}
}
// TestSignatureMemberIsExcluded checks that a document signs its content and
// not its own signature, so a signed document round-trips.
func TestSignatureMemberIsExcluded(t *testing.T) {
signer, pub, _ := GenerateKey("key-1")
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": pub})
d := newDoc("R-1", "stable")
sig, _ := signer.Sign(d)
attached := d
attached.Revision.Signature = &sig
if err := v.Verify(attached, &sig); err != nil {
t.Fatalf("a document carrying its own signature does not verify: %v", err)
}
}
func TestUnsignedIsRefused(t *testing.T) {
_, pub, _ := GenerateKey("key-1")
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": pub})
// A verifier that passes unsigned input is worse than no verifier, because
// it reports success.
if err := v.Verify(newDoc("R-1", "stable"), nil); !errors.Is(err, ErrUnsigned) {
t.Errorf("unsigned document returned %v, want ErrUnsigned", err)
}
}
func TestUntrustedKeyIsRefused(t *testing.T) {
rogue, _, _ := GenerateKey("rogue-key")
_, trustedPub, _ := GenerateKey("key-1")
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": trustedPub})
d := newDoc("R-1", "stable")
sig, _ := rogue.Sign(d)
if err := v.Verify(d, &sig); !errors.Is(err, ErrUnknownKey) {
t.Errorf("a signature from an untrusted key returned %v, want ErrUnknownKey", err)
}
}
func TestAlgorithmCannotBeDowngraded(t *testing.T) {
signer, pub, _ := GenerateKey("key-1")
v := NewVerifier(map[string]ed25519.PublicKey{"key-1": pub})
d := newDoc("R-1", "stable")
sig, _ := signer.Sign(d)
sig.Algorithm = "none"
if err := v.Verify(d, &sig); !errors.Is(err, ErrUnsupportedAlgorithm) {
t.Errorf("an algorithm downgrade returned %v, want ErrUnsupportedAlgorithm", err)
}
}
// TestCanonicalizationIsStable matters because an adapter or Daimon in another
// language must be able to produce a signature this verifier accepts.
func TestCanonicalizationIsStable(t *testing.T) {
d := newDoc("R-1", "stable")
first, err := Canonicalize(d)
if err != nil {
t.Fatal(err)
}
for i := 0; i < 100; i++ {
again, err := Canonicalize(d)
if err != nil {
t.Fatal(err)
}
if string(again) != string(first) {
t.Fatalf("canonical form varied between runs:\n%s\n%s", first, again)
}
}
// Key order in the source must not change the canonical bytes.
fromMap := map[string]any{
"revision": map[string]any{
"upstream": "http://adapter:8080",
"state": "stable",
"id": "R-1",
},
}
mapForm, err := Canonicalize(fromMap)
if err != nil {
t.Fatal(err)
}
if string(mapForm) != string(first) {
t.Errorf("canonical form depends on key order:\n%s\n%s", first, mapForm)
}
}
func TestNewSignerRejectsBadInput(t *testing.T) {
_, priv, _ := ed25519.GenerateKey(nil)
if _, err := NewSigner("", priv); err == nil {
t.Error("a signer with no key id was accepted")
}
if _, err := NewSigner("key-1", ed25519.PrivateKey("short")); err == nil {
t.Error("a malformed private key was accepted")
}
}

14
workplans/FLUID-WP-0004-publication-and-control-apis.md
View file

@ -4,7 +4,7 @@ type: workplan
title: "Revision publication pipeline and control APIs"
domain: infotech
repo: fluid-core
status: done
status: active
owner: worsch
topic_slug: fluid-core
created: "2026-09-04"
@ -26,7 +26,7 @@ bypass.
```task
id: FLUID-WP-0004-T01
status: done
status: todo
priority: high
state_hub_task_id: "74bf29c7-b8a6-5a18-bfc7-94641d12564b"
```
@ -38,7 +38,7 @@ CHECK, SIGN, PUBLISH, ROUTE. Each stage emits an audit event.
```task
id: FLUID-WP-0004-T02
status: done
status: todo
priority: high
state_hub_task_id: "9449aa0b-3c73-56d5-8768-df94b26f0282"
```
@ -51,7 +51,7 @@ tested.
```task
id: FLUID-WP-0004-T03
status: done
status: todo
priority: high
state_hub_task_id: "ba302369-3243-57ac-a34a-9a0a29002ac5"
```
@ -62,7 +62,7 @@ Blueprint §44.1: create, verify, publish, set state, query lineage.
```task
id: FLUID-WP-0004-T04
status: done
status: todo
priority: medium
state_hub_task_id: "27c597da-4f58-5a0d-b70e-8ba0749d9002"
```
@ -75,7 +75,7 @@ governed — the API records a proposal, it does not enact one.
```task
id: FLUID-WP-0004-T05
status: done
status: todo
priority: high
state_hub_task_id: "b2f96c6c-2c37-5892-bafd-ab9eb2c3f8c2"
```
@ -90,7 +90,7 @@ policy.
```task
id: FLUID-WP-0004-T06
status: done
status: todo
priority: medium
state_hub_task_id: "70cc234b-9570-5c51-8620-1a7d32b9a25c"
```