diff --git a/docs/changes/unreleased/1888-wormhole-two-computers.md b/docs/changes/unreleased/1888-wormhole-two-computers.md index 6784edd9e6..99be5c0ee5 100644 --- a/docs/changes/unreleased/1888-wormhole-two-computers.md +++ b/docs/changes/unreleased/1888-wormhole-two-computers.md @@ -61,5 +61,4 @@ invalidates: Pair only computers you trust: they receive your shared identity and sealed vault. Revoking a lost computer cannot take back its downloaded chats or keys; identity rotation reseals the chats under a replacement and retires the relay's old copy. -The hosted worker's tests isolate local state from the caller's home. Its staging-first -operator runbook is relay/hosted/DEPLOY.md; deployment is performed by the owner. +The hosted worker's tests isolate local state from the caller's home. diff --git a/docs/changes/unreleased/1920-tidy-relay-folder.md b/docs/changes/unreleased/1920-tidy-relay-folder.md new file mode 100644 index 0000000000..28b88e4944 --- /dev/null +++ b/docs/changes/unreleased/1920-tidy-relay-folder.md @@ -0,0 +1,6 @@ +--- +kind: changed +title: tidy relay examples and self-hosting defaults +pr: 1920 +surface: [remote, docs] +--- diff --git a/internal/relaybill/count.go b/internal/relaybill/count.go deleted file mode 100644 index fe396b06af..0000000000 --- a/internal/relaybill/count.go +++ /dev/null @@ -1,74 +0,0 @@ -package relaybill - -import ( - "net/http" - "sort" - "strings" - "sync" -) - -// Counter tallies the requests the client makes, by method and the first two -// path segments, so the billed numbers can be checked against what the client -// says it sent. It wraps a transport and changes nothing on the wire. -type Counter struct { - Next http.RoundTripper - mu sync.Mutex - n map[string]int -} - -// RoundTrip counts the request and passes it on. -func (c *Counter) RoundTrip(r *http.Request) (*http.Response, error) { - c.mu.Lock() - if c.n == nil { - c.n = map[string]int{} - } - c.n[kindOf(r)]++ - c.mu.Unlock() - return c.Next.RoundTrip(r) -} - -// kindOf names a request by what it does, not by the ids in its path. -func kindOf(r *http.Request) string { - parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/") - return r.Method + " /" + strings.Join(parts[:min(len(parts), 3)], "/") -} - -// Snapshot is the tally so far, and Since is the tally's growth from an earlier one. -func (c *Counter) Snapshot() map[string]int { - c.mu.Lock() - defer c.mu.Unlock() - out := make(map[string]int, len(c.n)) - for k, v := range c.n { - out[k] = v - } - return out -} - -// Since answers what was counted after the earlier snapshot, without zero rows. -func (c *Counter) Since(before map[string]int) map[string]int { - out := map[string]int{} - for k, v := range c.Snapshot() { - if d := v - before[k]; d > 0 { - out[k] = d - } - } - return out -} - -// Total is the sum of a tally. -func Total(n map[string]int) (sum int) { - for _, v := range n { - sum += v - } - return sum -} - -// Sorted lists a tally's kinds in order, for a stable report. -func Sorted(n map[string]int) []string { - keys := make([]string, 0, len(n)) - for k := range n { - keys = append(keys, k) - } - sort.Strings(keys) - return keys -} diff --git a/internal/relaybill/device.go b/internal/relaybill/device.go deleted file mode 100644 index 4bac3c34d7..0000000000 --- a/internal/relaybill/device.go +++ /dev/null @@ -1,122 +0,0 @@ -package relaybill - -import ( - "context" - "fmt" - "os" - "path/filepath" - "time" - - "github.com/Agent-Field/codeaf/internal/cell" - "github.com/Agent-Field/codeaf/internal/cellstore" - "github.com/Agent-Field/codeaf/internal/executor" - "github.com/Agent-Field/codeaf/internal/identity" - "github.com/Agent-Field/codeaf/internal/session" - "github.com/Agent-Field/codeaf/internal/syncsetup" -) - -// machine is one computer of the identity: its own home, engine data and disk. -type machine struct { - name string - sync *syncsetup.Sync - eng cellstore.Engine - roots string // where this machine keeps the chats it takes -} - -// newMachine makes a machine that holds the identity and nothing else, so a -// chat reaches it only through the relay. The first machine passes id nil and -// makes the identity; every later one adopts it. -func newMachine(dir, name, binary string, id *identity.Identity) (*machine, error) { - home := filepath.Join(dir, name, "home") - if err := os.MkdirAll(home, 0o700); err != nil { - return nil, err - } - if err := giveIdentity(home, id); err != nil { - return nil, err - } - s, ok, err := syncsetup.Open(home) - if err != nil || !ok { - return nil, fmt.Errorf("relaybill: open sync for %s: %v (on: %v)", name, err, ok) - } - eng := cellstore.EngineFor("") - eng.DataRoot, eng.Binary, eng.Transport = filepath.Join(dir, name, "data"), binary, cellstore.Spawn{Binary: binary} - return &machine{name: name, sync: s, eng: eng, roots: filepath.Join(dir, name, "roots")}, nil -} - -func giveIdentity(home string, id *identity.Identity) error { - if id == nil { - _, err := identity.Ensure(home) - return err - } - _, err := identity.Adopt(home, *id, false) - return err -} - -// continuer is this machine's take side, the one entry the screen calls. -func (m *machine) continuer() *syncsetup.Continuer { - return m.sync.Continuer(m.eng, syncsetup.TakeOptions{DeviceName: m.name, RootFor: func(id string) string { - return filepath.Join(m.roots, id) - }}) -} - -// open holds chat c on this machine the way a window does: the drive side -// publishes each sealed turn, beats the lease, and gives it back on Close. -func (m *machine) open(ctx context.Context, c cell.Cell, work string) (*openChat, error) { - eng := m.eng - eng.Workspace = work - drive, err := m.sync.Drive(ctx, eng, c, syncsetup.DriveOptions{DeviceName: m.name}) - if err != nil { - return nil, err - } - seat, err := cellstore.SeatOver(executor.HostBound, c, work, nil, nil, - func(cellstore.Engine) cellstore.Store { return drive.Store(eng) }) - if err != nil { - return nil, err - } - return &openChat{drive: drive, seat: executor.Gated(seat, drive.Gate), cell: c, work: work}, nil -} - -// workOf is the folder a taken chat works in. -func workOf(c cell.Cell) string { return session.Place{Dir: c.Root, Owned: true}.Work() } - -// openChat is one chat open on one machine. -type openChat struct { - drive *syncsetup.Drive - seat executor.Seat - cell cell.Cell - work string - calls int -} - -// say is one scripted model turn: the transcript grows by a line and a file is -// written, both sealed through the seat as the door seals a tool call. -func (o *openChat) say(ctx context.Context, text string) error { - o.calls++ - note := filepath.Join(o.work, fmt.Sprintf("turn-%d-%d.txt", time.Now().UnixNano(), o.calls)) - line := fmt.Sprintf(`{"type":"message","role":"assistant","content":%q}`+"\n", text) - if err := appendTo(session.Place{Dir: o.cell.Root}.Transcript(), line); err != nil { - return err - } - call := executor.Call{Tool: "bash", Args: []byte(`{"command":"echo"}`)} - return o.seat.Around(ctx, call, func() ([]byte, bool) { - return nil, os.WriteFile(note, []byte(text), 0o600) != nil - }) -} - -// close sends what is unsent and gives the lease back. -func (o *openChat) close(ctx context.Context) error { return o.drive.Close(ctx) } - -func appendTo(path, text string) error { - if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { - return err - } - f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o600) - if err != nil { - return err - } - if _, err := f.WriteString(text); err != nil { - _ = f.Close() - return err - } - return f.Close() -} diff --git a/internal/relaybill/doc.go b/internal/relaybill/doc.go deleted file mode 100644 index 40685dd57e..0000000000 --- a/internal/relaybill/doc.go +++ /dev/null @@ -1,16 +0,0 @@ -// Package relaybill drives the shape of a heavy hosted-relay user through the -// real client, so that what the relay bills for it can be read back from the -// Cloudflare analytics and scaled to a month. -// -// The shape is one identity with its home screen open, a chat held on one -// machine with its lease kept alive, and a number of moves of that chat: warm -// moves between two machines that already hold its objects and cold moves to a -// machine that has never seen it. Every request is made by the code the program -// ships (syncsetup's Drive and Continuer, the list read of the home screen), so -// a change in the client changes the bill this measures; nothing here speaks -// the wire by hand. -// -// The home screen's schedule is the one thing the caller supplies, because it -// lives in the terminal surface: [Schedule] is the pace the screen asks the -// directory at, and the surface's own test hands over its real one. -package relaybill diff --git a/internal/relaybill/project.go b/internal/relaybill/project.go deleted file mode 100644 index a8805053d6..0000000000 --- a/internal/relaybill/project.go +++ /dev/null @@ -1,51 +0,0 @@ -package relaybill - -import ( - "os" - "path/filepath" - "time" - - "github.com/Agent-Field/codeaf/internal/cell" - "github.com/Agent-Field/codeaf/internal/session" -) - -// smallRepo is what the moved chat works on: a few source files and one binary -// of a few kilobytes, the size of a person's first project. -func smallRepo(dir string) (string, error) { - blob := make([]byte, 9000) - for i := range blob { - blob[i] = byte(i * 31) - } - files := map[string][]byte{ - "README.md": []byte("# project\n"), - "src/main.go": []byte("package main\n\nfunc main() {}\n"), - "src/util.go": []byte("package main\n\nfunc util() int { return 1 }\n"), - "data/blob.bin": blob, - } - for path, body := range files { - full := filepath.Join(dir, path) - if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { - return "", err - } - if err := os.WriteFile(full, body, 0o644); err != nil { - return "", err - } - } - return dir, nil -} - -// newChat makes the chat cell the first machine starts with, carrying the -// session record and transcript header every real chat has so that a machine -// that takes it finds the project. -func newChat(root, work string) (cell.Cell, error) { - c, err := cell.CreateIn(root, cell.Options{Class: cell.HostBound}) - if err != nil { - return c, err - } - meta := session.Meta{ID: c.ID, Title: "heavy user", Workspace: work, Created: time.Now()} - if err := session.SaveMeta(c.Root, meta); err != nil { - return c, err - } - header := `{"type":"session","version":3,"id":"` + c.ID + `","cwd":"` + work + `","timestamp":"2026-09-29T10:00:00Z"}` + "\n" - return c, appendTo(session.Place{Dir: c.Root}.Transcript(), header) -} diff --git a/internal/relaybill/run.go b/internal/relaybill/run.go deleted file mode 100644 index 5743b2a4ee..0000000000 --- a/internal/relaybill/run.go +++ /dev/null @@ -1,222 +0,0 @@ -package relaybill - -import ( - "context" - "encoding/json" - "fmt" - "os" - "path/filepath" - "time" - - "github.com/Agent-Field/codeaf/internal/cell" - "github.com/Agent-Field/codeaf/internal/identity" -) - -// Config is one run of the heavy-user shape. -type Config struct { - Dir string // scratch folder for every machine's home, data and roots - Binary string // the engine program - Idle time.Duration // how long the home screen stays open with the lease held - Warm int // moves between two machines that already hold the chat - Cold int // moves to a machine that has never seen it - Gap time.Duration // quiet between phases, so each is read from the analytics on its own minutes - Tick time.Duration // the home screen's own beat - Schedule func() Schedule - Log func(format string, args ...any) -} - -// Phase is one span of the run the analytics are read for on its own. -type Phase struct { - Name string `json:"name"` - Start time.Time `json:"start"` - End time.Time `json:"end"` - Seconds float64 `json:"seconds,omitempty"` - Count int `json:"count,omitempty"` - // Steady says the phase is a rate to be scaled by its seconds, so the analytics read only its own minutes. - Steady bool `json:"steady,omitempty"` - Client map[string]int `json:"client_requests"` // what the client says it sent -} - -// Manifest is what the analytics reader is given: the window of each phase. -type Manifest struct { - URL string `json:"url"` - Script string `json:"script,omitempty"` - Start time.Time `json:"start"` - End time.Time `json:"end"` - Phases []Phase `json:"phases"` -} - -// run holds the state the phases share. -type run struct { - cfg Config - ctx context.Context - counter *Counter - id identity.Identity - first *machine - chat cell.Cell - holder *machine // the machine that holds the chat now - prior *machine // the machine that held it before - open *openChat - spare int // machines made so far - out Manifest -} - -// Run makes one identity, plays the shape through the real client against the -// relay named by CODEAF_SYNC_URL, and returns the window of each phase. The -// moves go first to prime two machines (unmeasured), then the idle hold, the -// warm moves and the cold moves, each in a window of its own. -func Run(ctx context.Context, cfg Config, url string, counter *Counter) (Manifest, error) { - r := &run{cfg: cfg, ctx: ctx, counter: counter, out: Manifest{URL: url}} - if err := r.setup(); err != nil { - return r.out, err - } - r.quiet() // the priming moves end in a minute of their own, apart from the idle hold's - steps := []func() (Phase, error){r.idle, r.warm, r.cold} - r.out.Start = time.Now().UTC() - for _, step := range steps { - p, err := step() - if err != nil { - return r.out, err - } - r.out.Phases = append(r.out.Phases, p) - r.quiet() - } - r.out.End = time.Now().UTC() - return r.out, r.release() -} - -// quiet waits out the gap between phases. -func (r *run) quiet() { - r.cfg.Log("quiet for %v", r.cfg.Gap) - select { - case <-r.ctx.Done(): - case <-time.After(r.cfg.Gap): - } -} - -// setup makes the first machine, the project and the chat, and primes the -// chat onto a second machine and a third so that the warm moves between those -// two find their objects already there. -func (r *run) setup() error { - first, err := newMachine(r.cfg.Dir, "m0", r.cfg.Binary, nil) - if err != nil { - return err - } - r.first, r.holder = first, first - r.id = first.sync.Identity - work, err := smallRepo(filepath.Join(r.cfg.Dir, "project")) - if err != nil { - return err - } - if r.chat, err = newChat(filepath.Join(r.cfg.Dir, "chats"), work); err != nil { - return err - } - if r.open, err = first.open(r.ctx, r.chat, work); err != nil { - return err - } - if err := r.open.say(r.ctx, "the first turn"); err != nil { - return err - } - for range 2 { - if _, err := r.move(r.freshMachine); err != nil { - return err - } - } - return nil -} - -// freshMachine is a machine that has never seen the chat. -func (r *run) freshMachine() (*machine, error) { - r.spare++ - return newMachine(r.cfg.Dir, fmt.Sprintf("m%d", r.spare), r.cfg.Binary, &r.id) -} - -// move is one move of the chat: the person looks at the list on the machine -// they are going to, the chat's holder gives it up with its turn sent, and -// that machine takes it and opens it. to supplies the machine. -func (r *run) move(to func() (*machine, error)) (*machine, error) { - dest, err := to() - if err != nil { - return nil, err - } - home := &Home{Src: dest.sync, Schedule: r.cfg.Schedule(), Tick: r.cfg.Tick} - if err := home.Read(r.ctx, time.Now()); err != nil { - return nil, fmt.Errorf("relaybill: list on %s: %w", dest.name, err) - } - if err := r.open.say(r.ctx, "a turn before the move"); err != nil { - return nil, err - } - if err := r.open.close(r.ctx); err != nil { - return nil, err - } - took, err := dest.continuer().Take(r.ctx, r.chat.ID) - if err != nil { - return nil, fmt.Errorf("relaybill: take on %s: %w", dest.name, err) - } - r.chat, r.prior, r.holder = took.Taken.Cell, r.holder, dest - r.open, err = dest.open(r.ctx, r.chat, workOf(r.chat)) - return dest, err -} - -// release gives the lease back at the end of the run. -func (r *run) release() error { return r.open.close(r.ctx) } - -// phase runs body and records its window and what the client sent in it. -func (r *run) phase(name string, body func() error) (Phase, error) { - before := r.counter.Snapshot() - p := Phase{Name: name, Start: time.Now().UTC()} - err := body() - p.End = time.Now().UTC() - p.Client = r.counter.Since(before) - r.cfg.Log("phase %s: %v, %d client requests", name, p.End.Sub(p.Start).Round(time.Second), Total(p.Client)) - return p, err -} - -// idle is the home screen open for cfg.Idle with the lease held and nothing -// being said: the screen's list reads and the drive side's beats. -func (r *run) idle() (Phase, error) { - p, err := r.phase("idle_hold", func() error { - ctx, stop := context.WithTimeout(r.ctx, r.cfg.Idle) - defer stop() - home := &Home{Src: r.holder.sync, Schedule: r.cfg.Schedule(), Tick: r.cfg.Tick} - home.Run(ctx) - return nil - }) - p.Seconds, p.Steady = p.End.Sub(p.Start).Seconds(), true - return p, err -} - -// warm is cfg.Warm moves, each back to the machine the chat just left: both -// were primed in setup, so each already holds the chat's objects. -func (r *run) warm() (Phase, error) { - p, err := r.phase("warm_moves", func() error { - return r.repeat(r.cfg.Warm, func() (*machine, error) { return r.prior, nil }) - }) - p.Count = r.cfg.Warm - return p, err -} - -// cold is cfg.Cold moves, each to a machine that has never seen the chat. -func (r *run) cold() (Phase, error) { - p, err := r.phase("cold_moves", func() error { return r.repeat(r.cfg.Cold, r.freshMachine) }) - p.Count = r.cfg.Cold - return p, err -} - -func (r *run) repeat(n int, to func() (*machine, error)) error { - for range n { - if _, err := r.move(to); err != nil { - return err - } - } - return nil -} - -// Write puts the manifest at path. -func (m Manifest) Write(path string) error { - raw, err := json.MarshalIndent(m, "", " ") - if err != nil { - return err - } - return os.WriteFile(path, append(raw, '\n'), 0o600) -} diff --git a/internal/relaybill/schedule.go b/internal/relaybill/schedule.go deleted file mode 100644 index f17465a45d..0000000000 --- a/internal/relaybill/schedule.go +++ /dev/null @@ -1,111 +0,0 @@ -package relaybill - -import ( - "context" - "time" - - "github.com/Agent-Field/codeaf/internal/chatlist" - "github.com/Agent-Field/codeaf/internal/dirwatch" -) - -// Schedule is when an open home screen asks the directory for its list: the -// surface's own pace, not a copy of it. -type Schedule interface { - // Ready says whether the wait since the last answer is over. - Ready(now time.Time) bool - // Landed files an answer: whether it differed from the one before, and - // whether the change socket was up when it came. - Landed(now time.Time, changed, watched bool) - // Hurry makes the next ask due now, as a person arriving on the screen does. - Hurry() -} - -// followable is a chat list source that can follow the directory's change feed. -type followable interface{ Follow() dirwatch.Follower } - -// Home is one open home screen: it reads the chat list through src whenever -// its schedule says so, looking at the clock every Tick, which is the screen's -// own beat, and follows the directory's change socket the way the screen does: -// a frame announcing a version past the last reading asks at once. -type Home struct { - Src chatlist.Source - Schedule Schedule - Tick time.Duration - last []chatlist.Row - version uint64 - feed dirwatch.Follower -} - -// Read is a person arriving on the screen: the socket is joined, the list is -// asked once at once, and the screen is left again. -func (h *Home) Read(ctx context.Context, now time.Time) error { - h.join() - defer h.leave() - h.Schedule.Hurry() - return h.ask(ctx, now) -} - -// Run keeps the screen open until ctx ends. An attended window is assumed: the -// bill being measured is the heavy user's, who is looking at it. -func (h *Home) Run(ctx context.Context) { - h.join() - defer h.leave() - h.Schedule.Hurry() - _ = h.ask(ctx, time.Now()) - tick := time.NewTicker(h.Tick) - defer tick.Stop() - for { - select { - case <-ctx.Done(): - return - case now := <-tick.C: - if h.Schedule.Ready(now) { - _ = h.ask(ctx, now) - } - case <-h.changes(): - if h.feedAhead() { - _ = h.ask(ctx, time.Now()) - } - } - } -} - -// join follows the feed if the source has one. -func (h *Home) join() { - if src, ok := h.Src.(followable); ok { - h.feed = src.Follow() - } -} - -// leave lets the feed go. -func (h *Home) leave() { - if h.feed != nil { - h.feed.Close() - h.feed = nil - } -} - -// changes is the feed's signal, or a channel that never fires without one. -func (h *Home) changes() <-chan struct{} { - if h.feed == nil { - return nil - } - return h.feed.Changes() -} - -// feedAhead says whether the feed has announced a version past the reading. -func (h *Home) feedAhead() bool { return h.feed != nil && h.feed.State().Version > h.version } - -// socketUp says whether the change socket is up. -func (h *Home) socketUp() bool { return h.feed != nil && h.feed.State().Up } - -// ask reads the list and tells the schedule whether it changed. A failed read -// counts as no change, as the screen counts it. -func (h *Home) ask(ctx context.Context, now time.Time) error { - rows, version, err := chatlist.RowsAt(ctx, h.Src) - h.Schedule.Landed(now, err == nil && !chatlist.Same(h.last, rows), h.socketUp()) - if err == nil { - h.last, h.version = rows, version - } - return err -} diff --git a/internal/tui3/relaybill_test.go b/internal/tui3/relaybill_test.go deleted file mode 100644 index cfd6857cbe..0000000000 --- a/internal/tui3/relaybill_test.go +++ /dev/null @@ -1,75 +0,0 @@ -//go:build relaybill - -package tui3 - -import ( - "context" - "flag" - "net/http" - "strings" - "testing" - "time" - - "github.com/Agent-Field/codeaf/internal/cell" - "github.com/Agent-Field/codeaf/internal/cell/celltest" - "github.com/Agent-Field/codeaf/internal/furrow" - "github.com/Agent-Field/codeaf/internal/home" - "github.com/Agent-Field/codeaf/internal/relaybill" - "github.com/Agent-Field/codeaf/internal/syncsetup" -) - -var ( - billURL = flag.String("bill-url", "", "base URL of the relay to load") - billOut = flag.String("bill-out", "", "where to write the run's manifest") - billIdle = flag.Duration("bill-idle", 20*time.Minute, "how long the home screen stays open with the lease held") - billWarm = flag.Int("bill-warm", 12, "moves between two machines that already hold the chat") - billCold = flag.Int("bill-cold", 1, "moves to a machine that has never seen the chat") - billGap = flag.Duration("bill-gap", 130*time.Second, "quiet between phases") -) - -// homePace is the home screen's own schedule for asking the directory (this -// package's machinePoll), offered to the load driver as the pace to ask at. -type homePace struct{ poll machinePoll } - -func (h *homePace) Ready(now time.Time) bool { return h.poll.ready(now) } -func (h *homePace) Landed(now time.Time, changed, watched bool) { - h.poll.landed(now, changed, watched) -} -func (h *homePace) Hurry() { h.poll.hurry() } - -// TestRelayBill plays one heavy user through the real client against the relay -// named by -bill-url, with this surface's real home pace, and writes the -// windows of its phases for the analytics reader. It is a measurement, not a -// check, so it only runs when asked for with the relaybill tag. -func TestRelayBill(t *testing.T) { - if *billURL == "" { - t.Fatal("-bill-url is required: the relay to load") - } - celltest.RequireEngine(t) - bin, err := furrow.ResolveOwned() - if err != nil { - t.Fatalf("no engine program: %v", err) - } - t.Setenv(home.EnvVar, t.TempDir()) - t.Setenv(syncsetup.URLVar, strings.TrimRight(*billURL, "/")) - t.Setenv(syncsetup.IntervalVar, "") - counter := &relaybill.Counter{Next: http.DefaultTransport} - http.DefaultTransport = counter - t.Cleanup(func() { http.DefaultTransport = counter.Next }) - - cfg := relaybill.Config{ - Dir: t.TempDir(), Binary: bin, Idle: *billIdle, Warm: *billWarm, Cold: *billCold, Gap: *billGap, - Tick: homeEvery, - Schedule: func() relaybill.Schedule { return &homePace{} }, - Log: func(f string, a ...any) { t.Logf(f, a...) }, - } - m, err := relaybill.Run(context.Background(), cfg, *billURL, counter) - if *billOut != "" { - if werr := m.Write(*billOut); werr != nil { - t.Fatal(werr) - } - } - if err != nil { - t.Fatal(err) - } -} diff --git a/relay/hosted/DEPLOY.md b/relay/hosted/DEPLOY.md deleted file mode 100644 index 3271e0c35c..0000000000 --- a/relay/hosted/DEPLOY.md +++ /dev/null @@ -1,241 +0,0 @@ -# Hosted relay: staging first, then production - -The owner performs account operations. This lane ran local Workers only: no login, -deployment, account request or production request. Commands marked **account only** -cannot be proved by `wrangler dev`. Even `deploy --dry-run` is left to the owner, -because the lane instructions forbid invoking `deploy`. - -## Prerequisites and permissions - -- Node 22.14 or newer (the local proof used 22.14; tests use `node:sqlite`), npm, the Go version in `go.mod`, - Python 3, curl and ripgrep. Use the lockfile's Wrangler, not a global installation. -- A Cloudflare account on Workers Paid, with R2 enabled, and the `agentfield.ai` - zone in that account. Confirm that the existing proxied DNS record for - `codeaf.agentfield.ai` is correct; this runbook does not replace it. -- An API token restricted to that account and zone. Deployment needs account - **Workers Scripts: Edit** (including Durable Object migrations), account - **Workers R2 Storage: Edit** for creating the bucket, zone **Workers Routes: Edit**, - and zone **Zone: Read** for resolving `zone_name`. Account discovery may also need - **Account Settings: Read**; set the account id explicitly. Billing queries need - account **Account Analytics: Read**, which may be a separate read-only token. - Do not paste tokens into files, arguments, shell history or this repository. -- Before an upgrade, retain the previous deployed version id and check that its - code can still read the current SQLite tables. A code rollback cannot undo data - migrations. The global budget uses the existing `NEWCOMERS` namespace; no new - Durable Object namespace is introduced by these limits. - -These are **separate config files**, not named Wrangler environments. Use -`-c staging.toml` and `-c production.toml`, without `--env staging`. - -| Config | Worker | R2 bucket | Public entry | -| --- | --- | --- | --- | -| `staging.toml` | `caf-relay-staging` | `caf-relay-staging-frames` | its account's workers.dev hostname, wire at root | -| `production.toml` | `caf-fabric` | `caf-fabric-frames` | `codeaf.agentfield.ai/fabric*` and `codeaf.agentfield.ai/p/*` | - -Both configs bind `IDENTITY`, `PAIR_GATE`, `MAILBOX`, `NEWCOMERS`, `LINK_GATE`, -`LINK_REQUEST` and `FRAMES`. Migrations `v1` and `v2` create the namespaces on the -first deploy. Production disables workers.dev, preview URLs and observability. -Staging enables the test proxy header and short timing/rate overrides; production -must keep `TRUST_PROXY` absent. Never copy staging's `CAF_LIMITS` wholesale into -production. `LEASE_POLICY=amended` stays the same on both. - -## Local rehearsal (executed) - -From `relay/hosted`: - -```sh -HOME=/proc sh -c 'npm ci --no-audit --no-fund && npm test' -sh test/deploy-local.sh -``` - -`HOME=/proc` is a read-only-home proof on Linux; ordinary `npm test` uses the same -hermetic wrapper on other systems. `test/local.sh` supplies temporary config, -cache, registry and log directories and disables Wrangler metrics, banners and -skills update prompts. It never repurposes the caller's home. Dependencies are -installed before the test; the test itself uses local tools. The project `.npmrc` -keeps npm's own cache and logs under the ignored `.wrangler/npm-cache` directory. - -`test/deploy-local.sh` starts the **production config locally**, with its own name, -R2 directory and `--inspector-port 0`, checks the `/p/k7m2q9xd` HTML page and a -`404 {"err":"not_found"}` beneath `/fabric`, then stops it. It starts a fresh local -Worker with small global cost ceilings and proves admission, storage, existing -reads, retries and owner erasure over signed HTTP. Finally it starts the **staging -config locally** through the canonical conformance runner, including the second -two-mailbox relay. Processes are cleaned up on exit; failed runs keep their temporary -evidence directory and print its path. This exercises the actual configs and wire; -DNS, account permissions, migrations on a remote account and billed usage remain -account-only checks. - -For a persistent local stand-in, the equivalent launch is: - -```sh -sh test/local.sh node_modules/.bin/wrangler dev --local -c production.toml \ - --name caf-production-local --ip 127.0.0.1 --port 8787 \ - --inspector-port 0 --persist-to /tmp/caf-production-local -``` - -The automated rehearsal executed that launch shape with an allocated free port, -a unique name and a temporary persistence directory. It ran these smoke checks -against its loopback URL; substitute the remote base only at the account step: - -```sh -curl -fsS http://127.0.0.1:8787/p/k7m2q9xd -curl -si http://127.0.0.1:8787/fabric/does-not-exist -``` - -For the complete two-relay matrix, from the repository root: - -```sh -GOFLAGS=-buildvcs=false GOMAXPROCS=4 \ - CODEAF_FURROW=/path/to/the/approved/furrow-engine \ - FURROW_ARTIFACT=/path/to/the/approved/furrow-engine \ - SKIP_E2E=1 bash scripts/relay-conformance.sh -``` - -This starts `bin/codeaf relay --listen --store ` -after `make build`, and two uniquely named local Workers, including a two-mailbox -Worker for `RelayFull`. The Go clients test both implementations. `SKIP_E2E` omits -only the separate long JS e2e script; `npm test` and all eight Go conformance -suites still run. `CONFORMANCE_WORK` chooses an evidence directory and -`CONFORMANCE_TRACKS=go` or `worker` selects a track for diagnosis. The local lane -needed explicit `-built` engine staging because the supplied engine did not match -the current pin; NOTES.md records that build limitation. Do not ship that override. - -Tracks run one after another by default, avoiding disk-heavy Go tests beside a -local Worker's timed requests. `CONFORMANCE_PARALLEL=1` opts into concurrency on -an isolated machine; `CONFORMANCE_SERIAL=1` explicitly forces sequential tracks. -`CONFORMANCE_CODEAF=/absolute/path/to/bin/codeaf` -uses a candidate already built with `make build`; `CONFORMANCE_LIVE_ONLY=1` then -runs the live Go wire track without repeating its in-process package tests. The -default matrix still includes those package tests and builds its own candidate. -`CONFORMANCE_WORKER_CONFIG` selects a config for the local Worker track; the -deployment rehearsal sets it to `staging.toml` and runs the eight suites separately. - -## Staging on the real account (account only; not executed) - -Use the owner's secure token environment and set `CLOUDFLARE_ACCOUNT_ID` to the -intended account. Inspect identity and the currently deployed version before writes: - -```sh -node_modules/.bin/wrangler whoami -node_modules/.bin/wrangler deployments list -c staging.toml -``` - -For the **first deployment only**, create the bucket; on an upgrade keep it: - -```sh -node_modules/.bin/wrangler r2 bucket create caf-relay-staging-frames -``` - -Check the bundle and bindings before publishing, then deploy staging: - -```sh -node_modules/.bin/wrangler deploy -c staging.toml --dry-run -node_modules/.bin/wrangler deploy -c staging.toml -``` - -Read the staging URL and version id from deployment output. Set `STAGING_URL` to -that workers.dev origin (no `/fabric` prefix); set `STAGING_VERSION` to the id. -Smoke the same static page and JSON refusal that the local rehearsal exercised: - -```sh -curl -fsS "$STAGING_URL/p/k7m2q9xd" -curl -si "$STAGING_URL/does-not-exist" -node_modules/.bin/wrangler deployments list -c staging.toml -node_modules/.bin/wrangler versions view "$STAGING_VERSION" -c staging.toml --json -``` - -Expect HTML with the strict CSP on the first request, and HTTP 404 with JSON -`{"err":"not_found"}` and `Codeaf-Now` on the second. From the repository root, -run the same tagged Go suite used by the local rehearsal: - -```sh -GOFLAGS=-buildvcs=false GOMAXPROCS=4 go test -p=2 -count=1 \ - -timeout 25m -tags relayurl ./internal/relayconf/ -relay-url="$STAGING_URL" -``` - -This creates real staging identities and ciphertext. Read every failure and skip. -`RelayFull` needs a second, two-mailbox relay and is skipped without -`-relay-small-url`; the complete local matrix covers it. Do not point that second -URL at production. Staging supplies the 4 s mailbox TTL, 2 s minimum erasure -grace and high per-address/request allowances the remaining cases need. - -## Production (account only; not executed) - -Proceed only after staging conformance is green. Record the previous production -version as `PREVIOUS_PRODUCTION_VERSION` before deployment. Create the production -bucket only if this is the first deployment: - -```sh -node_modules/.bin/wrangler deployments list -c production.toml -node_modules/.bin/wrangler r2 bucket create caf-fabric-frames -node_modules/.bin/wrangler deploy -c production.toml --dry-run -node_modules/.bin/wrangler deploy -c production.toml -``` - -The dry run must name the production bucket and both existing-host routes, -`/fabric*` and `/p/*`. It must not expose the wire at the hostname root. Record -the new version id as `PRODUCTION_VERSION`, then check the exact local smoke paths -on the production host: - -```sh -curl -fsS https://codeaf.agentfield.ai/p/k7m2q9xd -curl -si https://codeaf.agentfield.ai/fabric/does-not-exist -node_modules/.bin/wrangler deployments list -c production.toml -node_modules/.bin/wrangler versions view "$PRODUCTION_VERSION" -c production.toml --json -``` - -Expect the same HTML and JSON 404. Do not run the unrestricted conformance suite -against production: it deliberately exhausts quotas and creates many identities. - -## Cost settings, billing and rollback - -`src/limits.js` is the default limit table. `CAF_LIMITS` can override -`globalNewIdentitiesPerDay` (10,000) and `globalStoreBytes` (1 TiB). A zero value -closes the corresponding door. Changing these vars is an account operation and -follows the same staging/dry-run/deploy order; it needs no code change. See README -for rolling-day admission, crash-conservative storage reservations, initial R2 -inventory and erasure. Limits bound new stored data, not all possible request -charges. Inactive-identity expiry is off and has no knob in this version. - -Billing is **account only**. `cost/billed.py` reads Cloudflare's analytics API and -needs `Account Analytics: Read`; neither a local Worker nor a fabricated invoice -can prove an actual charge. Its help and 13 local Python tests were executed: - -```sh -python3 cost/billed.py --help -python3 -m unittest discover -s cost -p 'test_*.py' -``` - -With the billing token securely supplied and actual UTC start/end times: - -```sh -python3 cost/billed.py --account "$CLOUDFLARE_ACCOUNT_ID" \ - --script caf-relay-staging --bucket caf-relay-staging-frames \ - --from "$BILLING_FROM" --to "$BILLING_TO" --out /tmp/caf-staging-billed.json -python3 cost/billed.py --account "$CLOUDFLARE_ACCOUNT_ID" \ - --script caf-fabric --bucket caf-fabric-frames \ - --from "$BILLING_FROM" --to "$BILLING_TO" --out /tmp/caf-production-billed.json -``` - -Allow for analytics settling. The script reports usage for Worker requests/CPU, -Durable Object requests/duration/rows, and R2 operations; it is not a payment or a -change to the account. Its scope refusal explains the missing Analytics permission. -Keep the account's usage/billing dashboard beside these measurements. - -To roll back **code**, choose a previously deployed version compatible with the -current stored data, then use the real account command: - -```sh -node_modules/.bin/wrangler rollback "$PREVIOUS_PRODUCTION_VERSION" -c production.toml \ - --message "Restore the previous relay version" -node_modules/.bin/wrangler deployments list -c production.toml -``` - -Repeat the smoke checks. Wrangler's rollback, deployments-list and versions-view -command syntax was checked with local `--help`; the operations themselves were -not executed. Never delete either R2 bucket, the Worker, or any Durable Object -namespace, and never apply a migration with `deleted_classes` to roll back code. -Those objects hold people's data. Rolling back below the global-budget version -also removes its admission/storage protection; choose a compatible version or -retain the current brake implementation while fixing the fault. diff --git a/relay/hosted/README.md b/relay/hosted/README.md index 129e8b14c0..5cf9abb351 100644 --- a/relay/hosted/README.md +++ b/relay/hosted/README.md @@ -1,6 +1,6 @@ # Hosted relay -The relay of `internal/relayserve` (contract sections 3, 4.4, 6 and 18) as a Cloudflare Worker. +The relay of `internal/relayserve` as a Cloudflare Worker. The Go client holds both implementations to `internal/relayconf`; run the local matrix below before claiming conformance for a build. The relay sees only ciphertext and keeps no account table: an identity is the hash of its key. @@ -39,7 +39,7 @@ it, checks the device is not revoked, spends the caller's rate budget, and serve holding the objects of the longest prefix of the ids that fits in about 1 MiB, in request order, and stops at the first id the index does not hold (`404 not_found` only when the first is absent). It locates the ids in the object's SQLite and reads each frame it touches once, as one ranged R2 - read, so a takeover pays one Worker request and one R2 read per frame, not per object. + read, so a takeover uses one Worker request and one R2 read per frame, not per object. - A rid held with other bytes is `409 conflict`, and the frame stores none of its objects. Puts run one at a time inside the object, so two racing puts of one rid cannot both win. - `/v1/store/stats` is kept across evictions and restarts: counts are batched in memory and written @@ -54,15 +54,15 @@ Directory revocation is `POST /v1/dir/devices/{id}/revoke` (204; unknown id 404 caller itself 400 `self_revoke`; twice is fine). A revoked device is `401 revoked` on both wires, and a device record cannot set or clear its own flag. -Directory watch (contract 21) is `GET /v1/dir/watch`, a signed WebSocket upgrade on the same object +Directory watch is `GET /v1/dir/watch`, a signed WebSocket upgrade on the same object (`src/watch.js`). It replaces the home screen's polling: the object keeps a durable version of the directory (`dirver` in its SQLite), bumps it on every change a person could see (a heartbeat that only moves a lease's expiry is not one), and sends `{"v":N}` to every socket. It accepts with the hibernation API and answers the client's `ping` with the platform's auto-response, so an open, quiet socket wakes -nothing and bills nothing. `test/request-scoped.test.js` keeps the object free of timers (but `Has`'s +nothing. `test/request-scoped.test.js` keeps the object free of timers (but `Has`'s bounded wait) and of outbound connections. `maxWatchers` (default 1000) caps sockets per identity. -Rotation (contract 20) is `POST|GET /v1/identity/rotation` on the same object (`src/rotation.js`, a +Rotation is `POST|GET /v1/identity/rotation` on the same object (`src/rotation.js`, a port of `internal/directory/rotation.go`). The state is the `rotation` field of the identity record in the object's SQLite. The first device to freeze owns the rotation, any device may thaw a frozen identity, nothing thaws a retired one. Routes marked `write` in `src/routes.js` (every directory verb @@ -143,19 +143,18 @@ canonical names, so the stricter side is the safe one. | `MAILBOX` | Durable Object `Mailbox` | one pairing, one per nameplate | | `LINK_GATE` | Durable Object `LinkGate` (SQLite) | link-request codes, per-IP counters, one object | | `LINK_REQUEST` | Durable Object `LinkRequest` | one pending request, one per code | -| `LEASE_POLICY` | var | `amended` (default: 90 s, a publish renews; stage 1H decision 8.7) or `stage1` (the frozen 30 s law) | +| `LEASE_POLICY` | var | `amended` (default: 90 s, a publish renews) or `stage1` (30 s leases) | | `CAF_LIMITS` | var, optional | JSON that overrides the numbers below | -| `BASE_PATH` | var, optional | a path prefix every route is served beneath (`/fabric` in production); empty, the default, serves from the root. A request outside the prefix is `404 not_found` | +| `BASE_PATH` | var, optional | a path prefix every route is served beneath (for example, `/fabric`); empty, the default, serves from the root. A request outside the prefix is `404 not_found` | | `TRUST_PROXY` | var, optional | `1` takes the caller's network from the first `X-Forwarded-For` address (the self-hosted relay's `--trust-proxy`); for test rigs only, never production, where `CF-Connecting-IP` is the one truth | `wrangler.toml` declares all of them and the migration `v1` that creates the four classes as -SQLite-backed; migration `v2` adds the link classes. The relay needs Workers Paid: the Free plan allows 10 ms of CPU a request and 100,000 -requests a day. +SQLite-backed; migration `v2` adds the link classes. ## Base path -The hosted deployment shares its hostname with other things, so it is mounted at -`https://codeaf.agentfield.ai/fabric`. `BASE_PATH = "/fabric"` makes the Worker serve every route +A deployment can share a hostname by using a path prefix, such as +`https://relay.example.com/fabric`. `BASE_PATH = "/fabric"` makes the Worker serve every route (store, directory, the `/v1/dir/watch` upgrade, pairing, link requests, rotation) beneath that prefix. `src/base.js` holds the whole mechanism: the Worker strips the prefix once, before dispatch, and a request outside it is the usual 404. @@ -170,17 +169,17 @@ The Go relay (`internal/relayserve`) stays at the root of its host: it has no ba who wants one puts their own proxy in front, and the proxy must pass the path on unchanged (the client signed it), or serve the relay on a host of its own. -## Deploy +## Run your own -[DEPLOY.md](DEPLOY.md) is the staging-first operator runbook and its local proof. -Nothing deploys by itself. `staging.toml` names `caf-relay-staging` and -`caf-relay-staging-frames` on workers.dev. `production.toml` names `caf-fabric` and -`caf-fabric-frames`, with two routes on the shared host: `/fabric*` for the wire and -`/p/*` for the pairing page. The wire stays beneath `BASE_PATH=/fabric`; the page -lives outside that prefix. Observability is off in production. +Copy `wrangler.toml` to a config of your own and choose a Worker name and R2 bucket name. +Create the bucket and deploy from `relay/hosted`: -The rollback keeps the bucket and Durable Object namespaces. Removing a route is -not a data rollback, and deleting storage is never part of rolling back code. +```sh +cp wrangler.toml wrangler.custom.toml +# Edit wrangler.custom.toml to match the bucket name below. +npx wrangler r2 bucket create your-relay-frames +npx wrangler deploy --config wrangler.custom.toml +``` ## Limits (`src/limits.js`, one table) @@ -207,64 +206,17 @@ Stored quotas (bytes, objects, frames a day), the stats counts and the per-IP co eviction. A frame the store already holds costs no quota, so a client that resends is never refused for it. The per-minute rates are kept in memory: an eviction can only forgive a burst. -## Cost note: Durable Object SQLite rows written - -DO SQLite bills rows written at $1 per million after 50 million a month included (rows read are -25 billion included and not a concern here). Every table in the object is `WITHOUT ROWID` or keyed -by the rowid, so a row is one row written, not a row and an index entry (an assumption to confirm -against billed usage on a real account; the stage 1H staging read-back is the method). - -| Operation | Rows written | -|---|---| -| put of a frame with n objects, new | n + 2 (the frame, each object, the usage row) | -| put of a frame already held | 0 | -| directory create, acquire, heartbeat, publish, release | 1 | -| stats | 1 per flush window (5 s) of store activity, not per request | -| `Has`, `Get`, list, cell read | 0 | -| an identity's first request | 1 (admitted) | - -For the session shape measured in the stage 1H decision (232 one-object puts, 197 directory writes) -that is about 930 rows a session; 10,000 identities at 8 sessions a month is 74M rows, so about -$24 a month over the included 50M. With frames batched to about 12 puts of 20 objects a session it -is about 510 rows, 41M a month, inside the included amount. Object count no longer costs memory, so -the only price of a large repository is these rows: 170,000 objects is 170,000 rows once, about -$0.17. - -## The other option - -The R2-only directory (a Worker with no Durable Object) was built and measured in the stage 1H -spike and is kept in git history, not in this tree: `git log wip/layout-v0-hspike -- relay/hosted/src/r2dir.js`. -It costs 1.3 to 2.3 times as much and a directory write takes tens of times longer, and it cannot -keep the in-flight-put rule by identity. - -## Deployment-wide cost brakes and data erasure - -`CAF_LIMITS` also accepts `globalNewIdentitiesPerDay` (default 10,000) and -`globalStoreBytes` (default 1 TiB, 1,099,511,627,776 bytes). Set either to `0` to close -that door. For example, merge this JSON into the chosen config's existing `CAF_LIMITS` -value, retaining staging's short test timings: - -```toml -CAF_LIMITS = '{"globalNewIdentitiesPerDay":1000,"globalStoreBytes":107374182400}' -``` +## Deployment limits and data erasure + +`CAF_LIMITS` optionally sets `globalNewIdentitiesPerDay` and `globalStoreBytes`. +Both are unlimited by default. Set either to a finite number to enable that limit; +`0` closes new admission or storage respectively. Per-identity and per-network limits +still apply. -A new identity over the global daily admission ceiling gets `429 relay_admission_full` -and `Retry-After` in seconds. Its rolling day starts with the first successful admission; -per-network admission still applies. An identity already admitted is never charged again. -New identities at the storage ceiling, and new frames that would exceed it, get -`507 relay_storage_full`. Existing data remains readable, a held frame can be retried, -and freeze/retire erasure remains available. These are deployment settings; the operator -can change them without editing code. Apply changed vars with the owner's normal -staging-first deployment procedure. - -The existing `NewcomerGate` keeps the global budget in SQLite. Its first use inventories -all existing R2 objects, page by page, before allowing new writes, so an upgrade includes -old stored bytes. Interrupted inventories repeat safely. A new frame reserves its bytes -before R2 is written, and retries share that reservation. A failed or interrupted write -may retain a conservative reservation; the owner's completed identity erasure returns -all its capacity. The extra budget rows are metadata; the byte ceiling measures R2 frame -bytes, not total Cloudflare spend. Existing per-identity, mailbox and link caps still apply. -Monitor request, database and storage charges as described in DEPLOY.md. +A new identity over the daily admission limit gets `429 relay_admission_full` with +`Retry-After`. New identities at the storage limit, and new frames that would exceed +it, get `507 relay_storage_full`. Existing data stays readable, held frames can be +retried, and identity erasure remains available. Every frame is under an identity prefix. Its owner can retire the identity through the signed rotation route (normally `codeaf identity rotate`); after the grace period, an diff --git a/relay/hosted/cost/bill.py b/relay/hosted/cost/bill.py deleted file mode 100644 index d2b540fb7f..0000000000 --- a/relay/hosted/cost/bill.py +++ /dev/null @@ -1,224 +0,0 @@ -#!/usr/bin/env python3 -"""Turn measured billed phase usage into monthly dollars per user, and render section 9 of STAGE-1H-DECISION.md. - -Arithmetic (every number comes from the run, the SHAPE block below, or model.PRICE): - background The idle_hold phase is a steady state: one home screen open, one lease held, idle heartbeats. - Its billed usage divided by its seconds is a per-second rate; times - home_hours_per_day * 3600 * days_per_month seconds it is the background month. - warm move warm_moves phase usage divided by its count. - cold move cold_moves phase usage divided by its count. - month background + warm_per_month * warm + cold_per_month * cold. - The result is a usage vector per user-month with the same keys as a billed phase. -Dollars: the per-user vector times N users, then charged with model.py's tier arithmetic (`over`, PRICE): - Workers base once, Worker requests and CPU ms, DO requests, DO GB-s, DO rows, R2 Class A and B. - Rows read and rows written are separate meters (ROWS_READ, ROWS_WRITTEN). - No storage line is billed here; storage is in model.py's common(). -Marginal dollars (the per-user column) use no free tier and no Workers base, the cost of one more user. -""" -import argparse -import json -import re -import sys - -from model import PRICE, over - -# One place for the assumed heavy-user shape; the manifest's "shape" and the command line override it. -# The plan's heavy user: a home screen open 8 h on each of 22 working days, and 40 warm plus 4 cold moves in the MONTH. -SHAPE = dict(home_hours_per_day=8, days_per_month=22, warm_per_month=40, cold_per_month=4) -# Durable Object SQLite rows, from developers.cloudflare.com/durable-objects/platform/pricing (fetched 2026-09-30): -# reads are a thousand times cheaper than writes, so they are two meters and not the one PRICE has. -ROWS_READ = dict(free=25e9, per_m=0.001) -ROWS_WRITTEN = dict(free=50e6, per_m=1.0) -SCENARIOS = [(10_000, 0.25), (10_000, 1.0), (100_000, 0.25), (100_000, 1.0)] -KEYS = ["worker_requests", "worker_cpu_ms", "do_requests", "do_active_s", "do_gb_s", - "do_rows_read", "do_rows_written", "r2_class_a", "r2_class_b"] -USAGE_ROWS = [("Worker requests", "worker_requests"), ("Worker CPU ms", "worker_cpu_ms"), - ("DO requests", "do_requests"), ("DO GB-s", "do_gb_s"), - ("DO rows read", "do_rows_read"), ("DO rows written", "do_rows_written"), - ("R2 Class A", "r2_class_a"), ("R2 Class B", "r2_class_b")] -PENDING = "n/a (pending)" -HEARTBEAT_S = 30 # directory.HeartbeatEvery: an idle held lease sends one directory write this often -HEADING = "## 9. Billed heavy-user run" - - -def scale(vec, k): - return {key: vec.get(key, 0) * k for key in KEYS} - - -def add(*vecs): - return {key: sum(v.get(key, 0) for v in vecs) for key in KEYS} - - -def class_a_floor(used, phase): - """A phase's billed usage with R2 Class A raised to the frame puts the client sent, each a PutObject.""" - puts = phase.get("client_requests", {}).get("POST /v1/store/frames", 0) - return {**used, "r2_class_a": max(used.get("r2_class_a", 0), puts)} - - -def month_usage(manifest, billed, shape): - """Per-user month as a usage vector, from the phase windows and the billed usage of each phase.""" - phases = {p["name"]: p for p in manifest["phases"]} - used = {n: class_a_floor(v, phases[n]) for n, v in billed["phases"].items()} - days = shape["days_per_month"] - background_s = shape["home_hours_per_day"] * 3600 * days - background = scale(used["idle_hold"], background_s / phases["idle_hold"]["seconds"]) - warm = scale(used["warm_moves"], shape["warm_per_month"] / phases["warm_moves"]["count"]) - cold = scale(used["cold_moves"], shape["cold_per_month"] / phases["cold_moves"]["count"]) - return add(background, warm, cold) - - -def meters(u, free, base): - """One row per priced meter: (name, quantity, price text, dollars). free=0 and base=0 is the marginal cost.""" - p = PRICE - def meter(name, key, qty, free_qty, per_m, unit): - return (name, qty, f"${per_m:g} per million {unit} after {free_qty * free:,.0f}", over(qty, free_qty * free, per_m)) - return [ - ("Workers base", base, "$5 a month", base * p["workers_base"]), - meter("Worker requests", "req", u["worker_requests"], p["req_free"], p["req_per_m"], "requests"), - meter("Worker CPU ms", "cpu", u["worker_cpu_ms"], p["cpu_free_ms"], p["cpu_per_m_ms"], "ms"), - meter("DO requests", "doreq", u["do_requests"], p["do_req_free"], p["do_req_per_m"], "requests"), - meter("DO GB-s", "gbs", u["do_gb_s"], p["do_gbs_free"], p["do_gbs_per_m"], "GB-s"), - meter("DO rows read", "rr", u["do_rows_read"], ROWS_READ["free"], ROWS_READ["per_m"], "rows"), - meter("DO rows written", "rw", u["do_rows_written"], ROWS_WRITTEN["free"], ROWS_WRITTEN["per_m"], "rows"), - meter("R2 Class A", "a", u["r2_class_a"], p["r2_a_free"], p["r2_a_per_m"], "ops"), - meter("R2 Class B", "b", u["r2_class_b"], p["r2_b_free"], p["r2_b_per_m"], "ops"), - ] - - -def charge(u, free, base): - """Dollars for one usage vector.""" - return sum(m[3] for m in meters(u, free, base)) - - -def breakdown(user_month, n, active, title): - """The markdown table of every meter for one scenario, so its total can be added by eye.""" - rows = meters(scale(user_month, n * active), 1, 1) - lines = [f"Dollar breakdown, {title}: {n:,} users, {active:.0%} active (free tiers and Workers base applied).", "", - "| Meter | Monthly quantity | Price | Dollars |", "|---|---:|---|---:|"] - lines += [f"| {name} | {num(q)} | {price} | ${d:,.2f} |" for name, q, price, d in rows] - return lines + [f"| **Total** | | | **${sum(r[3] for r in rows):,.2f}** |"] - - -def dollars(user_month): - """Marginal per user-month, then each scenario with free tiers and the Workers base applied.""" - out = [("Per user-month (marginal, no free tier)", charge(user_month, 0, 0))] - for n, active in SCENARIOS: - out.append((f"{n:,} users, {active:.0%} active", charge(scale(user_month, n * active), 1, 1))) - return out - - -def num(x): - return f"{x:,.0f}" if abs(x) >= 100 else f"{x:,.2f}" - - -def table(head, titles, rows): - """A markdown table with one column per run; rows are (name, [cell per run]).""" - lines = ["| " + " | ".join([head] + titles) + " |", "|---|" + "---:|" * len(titles)] - return lines + ["| " + " | ".join([name] + cells) + " |" for name, cells in rows] - - -def usage_rows(months): - return [(label, [num(m[k]) for m in months]) for label, k in USAGE_ROWS] - - -def dollar_rows(months): - all_dollars = [dollars(m) for m in months] - return [(all_dollars[0][i][0], [f"${d[i][1]:,.2f}" if d[i][1] < 100 else f"${d[i][1]:,.0f}" for d in all_dollars]) - for i in range(len(all_dollars[0]))] - - -def ratio_lines(titles, months): - """Each later column against the first: the two meters the push work is meant to move.""" - first, out = months[0], [] - for title, m in zip(titles[1:], months[1:]): - parts = [f"{label} {m[k] / first[k]:.2f}" for label, k in (("DO GB-s", "do_gb_s"), ("DO requests", "do_requests"), - ("DO rows written", "do_rows_written")) if first[k]] - out.append(f"Ratio {title} / {titles[0]} per user-month: " + ", ".join(parts) + ".") - return out - - -def rows_sentence(manifest, billed): - """Which verb writes the rows: the idle hold sends only heartbeats and list reads, and only heartbeats write.""" - idle = next(p for p in manifest["phases"] if p["name"] == "idle_hold") - beats = idle["seconds"] / HEARTBEAT_S - written = billed["phases"]["idle_hold"]["do_rows_written"] - return (f"Rows written are the largest line because a held lease heartbeats every {HEARTBEAT_S} s for the whole time the " - f"home is open: in the idle hold (about {beats:.0f} heartbeats, nothing else writes) the relay wrote {written:,.0f} " - f"rows, {written / beats:.1f} per heartbeat. In the Worker (`directory.js`) a heartbeat runs one `INSERT OR REPLACE INTO dir`, whose measured cost fits a replace " - "counted as a delete and an insert, and the `UPDATE dirver` of the durable version only when the list-visible record changed, " - "which a beat that moves the expiry alone does not. Rows read come from the list, which selects every directory row.") - - -def intro(columns, shape): - """The shared paragraph: the shape, the arithmetic, then each run's window and notes.""" - s = shape - text = (f"Shape per user: home screen open {s['home_hours_per_day']} h/day on {s['days_per_month']:g} days a month, and " - f"{s['warm_per_month']:g} warm plus {s['cold_per_month']:g} cold moves in the month. The idle_hold phase rate per " - "second times the open-home seconds is the background month; each move type adds its phase cost per move times its " - "monthly count. Dollars charge the usage times N users with the free tiers and Workers base in model.py; storage is " - "not billed here. R2 Class A is the larger of the billed count and the frame puts the client sent, because the " - "analytics were seen to miss puts (12 frame puts, 0 billed).") - lines = [text, "", rows_sentence(columns[0]["manifest"], columns[0]["billed"]), ""] - for c in columns: - m = c["manifest"] - notes = " ".join([m.get("note", "")] + [p["note"].capitalize() + "." for p in m["phases"] if p.get("note")]).strip() - lines.append(f"- **{c['title']}**: `{c['billed']['script']}`, window {m['start']} to {m['end']}. {notes}".rstrip()) - return "\n".join(lines) - - -def render(columns, shape): - """Section 9 for any number of runs, each a dict of title, manifest and billed.""" - months = [month_usage(c["manifest"], c["billed"], shape) for c in columns] - titles = [c["title"] for c in columns] - lines = [HEADING, "", intro(columns, shape), "", "Usage per user-month.", ""] - lines += table("Usage", titles, usage_rows(months)) - lines += ["", "Dollars per month.", ""] - lines += table("Scenario", titles, dollar_rows(months)) - for title, m in zip(titles, months): - lines += [""] + breakdown(m, 10_000, 1.0, title) - lines += [""] + ratio_lines(titles, months) - return "\n".join(lines).rstrip() + "\n" - - -def splice(doc, section): - """Replace section 9 (heading to the next '## ' heading or the end) or append it; no other byte changes.""" - m = re.search(r"^## 9\. Billed heavy-user run[^\n]*\n", doc, re.M) - if not m: - return doc.rstrip("\n") + "\n\n" + section if doc else section - nxt = re.compile(r"^## ", re.M).search(doc, m.end()) - end = nxt.start() if nxt else len(doc) - old = doc[m.start():end] - tail = old[len(old.rstrip("\n")):] or "\n" - return doc[:m.start()] + section.rstrip("\n") + tail + doc[end:] - - -def load(path): - with open(path) as f: - return json.load(f) - - -def shape_for(manifest, overrides): - return {**SHAPE, **manifest.get("shape", {}), **{k: v for k, v in overrides.items() if v is not None}} - - -def main(argv=None): - ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) - ap.add_argument("--column", nargs=3, action="append", required=True, metavar=("TITLE", "MANIFEST", "BILLED"), - help="one run, in the order the columns are shown; repeat for each run") - ap.add_argument("--doc") - for key in SHAPE: - ap.add_argument("--" + key.replace("_", "-"), dest=key, type=float) - a = ap.parse_args(argv) - columns = [dict(title=t, manifest=load(m), billed=load(b)) for t, m, b in a.column] - section = render(columns, shape_for(columns[0]["manifest"], {k: getattr(a, k) for k in SHAPE})) - if not a.doc: - sys.stdout.write(section) - return - with open(a.doc) as f: - doc = f.read() - with open(a.doc, "w") as f: - f.write(splice(doc, section)) - - -if __name__ == "__main__": - main() diff --git a/relay/hosted/cost/billed.py b/relay/hosted/cost/billed.py deleted file mode 100644 index 6f0645e7c0..0000000000 --- a/relay/hosted/cost/billed.py +++ /dev/null @@ -1,382 +0,0 @@ -#!/usr/bin/env python3 -"""Read what Cloudflare billed a Worker and its R2 bucket for a time window. - -The numbers come from the GraphQL analytics API, which is the same source the invoice uses, so a -load run can be priced from measured usage instead of estimates. Standard library only. - -Units of the API fields this file reads (checked with __type introspection on 2026-09-30): - workersInvocationsAdaptive.sum.cpuTimeUs microseconds of CPU -> worker_cpu_ms = /1000 - durableObjectsPeriodicGroups.sum.activeTime microseconds active -> do_active_s = /1e6 - durableObjectsPeriodicGroups.sum.duration GB*s, already as billed -> do_gb_s - (check: STAGE-1H-DECISION.md 8.8 shows 1.83 GB-s over 14.3 s active, i.e. 128 MB x 14.3 s) - durableObjectsPeriodicGroups.sum.rowsRead/rowsWritten rows (SQLite-backed objects) - -Exit codes: 0 ok, 2 bad usage or no data, 3 token lacks the Analytics scope, 4 API or network failure. -""" -import argparse -import json -import os -import re -import subprocess -import sys -import time -import urllib.error -import urllib.request -from datetime import datetime, timedelta, timezone - -ENDPOINT = "https://api.cloudflare.com/client/v4/graphql" -DEFAULT_ACCOUNT = "190bd4aa38c975a1ae33f2521f80d01a" -DEFAULT_SCRIPT = "caf-relay-staging" -DEFAULT_BUCKET = "caf-relay-staging-frames" -WRANGLER_CONFIG = "~/.config/.wrangler/config/default.toml" -WRANGLER_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") - -# The one classification table, from Cloudflare's published R2 pricing categories -# (developers.cloudflare.com/r2/pricing, fetched 2026-09-30). Free operations map to None. -R2_CLASS = { - **dict.fromkeys( - ["ListBuckets", "PutBucket", "ListObjects", "PutObject", "CopyObject", - "CompleteMultipartUpload", "CreateMultipartUpload", "LifecycleStorageTierTransition", - "ListMultipartUploads", "UploadPart", "UploadPartCopy", "ListParts", - "PutBucketEncryption", "PutBucketCors", "PutBucketLifecycleConfiguration"], "A"), - **dict.fromkeys( - ["HeadBucket", "HeadObject", "GetObject", "UsageSummary", "GetBucketEncryption", - "GetBucketLocation", "GetBucketCors", "GetBucketLifecycleConfiguration"], "B"), - **dict.fromkeys(["DeleteObject", "DeleteBucket", "AbortMultipartUpload"], None), -} -# Cloudflare bills every request except unauthorized ones, so a miss (notFound) is a billed -# Class B read. We count success and notFound and leave out error statuses, which are rare here. -BILLED_STATUSES = {"success", "notFound"} - -METRICS = ["worker_requests", "worker_cpu_ms", "do_requests", "do_active_s", "do_gb_s", - "do_rows_read", "do_rows_written", "r2_class_a", "r2_class_b"] - -SCOPE_FIX = ("the token lacks the Account Analytics Read scope.\n" - "Fix: create an API token with permission 'Account Analytics:Read' " - "(dash.cloudflare.com/profile/api-tokens) and run: export CLOUDFLARE_API_TOKEN=") - - -class ScopeError(Exception): - """The API refused the query for lack of the analytics scope.""" - - -class AuthExpired(Exception): - """The API answered 401, so the token is expired or wrong.""" - - -# ---------- time helpers ---------- - -def parse_time(text): - return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc) - - -def iso(moment): - return moment.strftime("%Y-%m-%dT%H:%M:%SZ") - - -def minute_window(start, end): - """Widen a window to whole minutes, because analytics are bucketed per minute and a - partial minute could not be separated from its neighbours anyway.""" - lo = start.replace(second=0, microsecond=0) - hi = end.replace(second=0, microsecond=0) - if hi < end: - hi += timedelta(minutes=1) - return lo, hi - - -# ---------- token and transport ---------- - -def token_from_wrangler(): - text = open(os.path.expanduser(WRANGLER_CONFIG)).read() - return re.search(r'oauth_token\s*=\s*"([^"]+)"', text).group(1) - - -def load_token(): - """An explicit API token wins, because it is the owner's deliberate grant of the scope.""" - return os.environ.get("CLOUDFLARE_API_TOKEN") or token_from_wrangler() - - -def refresh_wrangler_token(): - """`wrangler whoami` renews an expired OAuth token as a side effect; nothing else is changed.""" - subprocess.run(["npx", "wrangler", "whoami"], cwd=WRANGLER_DIR, check=False, - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=180) - - -def post(query, token): - body = json.dumps({"query": query}).encode() - headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} - try: - with urllib.request.urlopen(urllib.request.Request(ENDPOINT, body, headers), timeout=60) as r: - return json.load(r) - except urllib.error.HTTPError as err: - if err.code == 401: - raise AuthExpired() - if err.code == 403: - raise ScopeError() - raise - - -def raise_on_errors(answer): - errors = answer.get("errors") or [] - text = json.dumps(errors).lower() - if errors and ("authz" in text or "not authorized" in text or "does not have access" in text): - raise ScopeError() - if errors: - raise RuntimeError("analytics API error: " + json.dumps(errors)[:400]) - - -class Client: - """Runs queries, refreshing the wrangler token once when it has expired.""" - - def __init__(self, account): - self.account = account - self.token = load_token() - self.refreshed = False - - def run(self, inner): - query = '{ viewer { accounts(filter:{accountTag:"%s"}) { %s } } }' % (self.account, inner) - try: - answer = post(query, self.token) - except AuthExpired: - if self.refreshed or os.environ.get("CLOUDFLARE_API_TOKEN"): - raise - self.refreshed = True - refresh_wrangler_token() - self.token = load_token() - return self.run(inner) - raise_on_errors(answer) - return answer["data"]["viewer"]["accounts"][0] - - -# ---------- queries ---------- - -def q_workers(script, lo, hi): - return ('workersInvocationsAdaptive(limit:10, filter:{scriptName:"%s", datetime_geq:"%s", ' - 'datetime_leq:"%s"}) { sum { requests cpuTimeUs } }' % (script, iso(lo), iso(hi))) - - -def q_do_groups(script, lo, hi): - return ('durableObjectsInvocationsAdaptiveGroups(limit:100, filter:{scriptName:"%s", ' - 'datetime_geq:"%s", datetime_leq:"%s"}) { dimensions { namespaceId } sum { requests } }' - % (script, iso(lo), iso(hi))) - - -def q_do_periodic(namespaces, lo, last_minute): - ids = json.dumps(sorted(namespaces)) - return ('durableObjectsPeriodicGroups(limit:1000, filter:{namespaceId_in:%s, ' - 'datetimeMinute_geq:"%s", datetimeMinute_leq:"%s"}) { dimensions { datetimeMinute } ' - 'sum { activeTime duration rowsRead rowsWritten } }' % (ids, iso(lo), iso(last_minute))) - - -def q_r2(bucket, lo, hi): - return ('r2OperationsAdaptiveGroups(limit:200, filter:{bucketName:"%s", datetime_geq:"%s", ' - 'datetime_leq:"%s"}) { dimensions { actionType actionStatus } sum { requests } }' - % (bucket, iso(lo), iso(hi))) - - -# ---------- parsing (pure) ---------- - -def parse_workers(rows): - requests = sum(r["sum"]["requests"] for r in rows) - cpu_us = sum(r["sum"]["cpuTimeUs"] for r in rows) - return {"worker_requests": int(requests), "worker_cpu_ms": cpu_us / 1000.0} - - -def parse_do_invocations(rows): - return int(sum(r["sum"]["requests"] for r in rows)) - - -def parse_do_periodic(rows): - def total(key): - return sum(r["sum"][key] for r in rows) - return {"do_active_s": total("activeTime") / 1e6, "do_gb_s": float(total("duration")), - "do_rows_read": int(total("rowsRead")), "do_rows_written": int(total("rowsWritten"))} - - -def classify(action_type, status): - """Return 'A', 'B' or None (free, unbilled or unknown) for one R2 operation row.""" - if status not in BILLED_STATUSES: - return None - return R2_CLASS.get(action_type) - - -def parse_r2(rows): - counts = {"A": 0, "B": 0} - for r in rows: - kind = classify(r["dimensions"]["actionType"], r["dimensions"]["actionStatus"]) - if kind: - counts[kind] += int(r["sum"]["requests"]) - return {"r2_class_a": counts["A"], "r2_class_b": counts["B"]} - - -def unknown_actions(rows): - return sorted({r["dimensions"]["actionType"] for r in rows} - set(R2_CLASS)) - - -def sum_phases(phases): - return {m: sum(p[m] for p in phases.values()) for m in METRICS} - - -# ---------- reading one phase ---------- - -def periodic_ranges(phases, steady=()): - """The minutes of the periodic Durable Object dataset that belong to each phase. - - That dataset is stamped when the runtime reports, not when work ran, and was seen up to a minute - off (a move at 21:10:04 showed its rows in the 21:09 bucket). So a phase owns every minute up to - the midpoint of the quiet gap that follows it, which is wide enough (two minutes or more) that - the shift cannot reach the next phase's work. A steady phase (an idle hold, whose rate is what is - wanted) owns exactly its own minutes: the quiet after it still holds the lease and would be counted - in rows that its seconds do not include.""" - order = sorted(phases.items(), key=lambda kv: kv[1][0]) - ranges, first = {}, None - for i, (name, (start, end)) in enumerate(order): - first = first or start.replace(second=0, microsecond=0) - if i + 1 < len(order): - mid = end + (order[i + 1][1][0] - end) / 2 - last = mid.replace(second=0, microsecond=0) - else: - last = (end + timedelta(minutes=1)).replace(second=0, microsecond=0) - if name in steady: - last = min(last, minute_window(start, end)[1] - timedelta(minutes=1)) - first = max(first, minute_window(start, end)[0]) - ranges[name] = (first, last) - first = last + timedelta(minutes=1) - return ranges - - -def read_phase(client, script, bucket, namespaces, start, end, minutes=None): - lo, hi = minute_window(start, end) - hi_event = hi - timedelta(seconds=1) - first_minute, last_minute = minutes or (lo, hi - timedelta(minutes=1)) - out = parse_workers(client.run(q_workers(script, lo, hi_event))["workersInvocationsAdaptive"]) - groups = client.run(q_do_groups(script, lo, hi_event))["durableObjectsInvocationsAdaptiveGroups"] - out["do_requests"] = parse_do_invocations(groups) - # The periodic dataset has no script name, so it is read per namespace found above and in - # the run-wide discovery (an idle namespace still bills duration in a phase without requests). - ids = set(namespaces) | {g["dimensions"]["namespaceId"] for g in groups} - periodic = [] - if ids: - periodic = client.run(q_do_periodic(ids, first_minute, last_minute))["durableObjectsPeriodicGroups"] - out.update(parse_do_periodic(periodic)) - r2_rows = client.run(q_r2(bucket, lo, hi_event))["r2OperationsAdaptiveGroups"] - bad = unknown_actions(r2_rows) - if bad: - print("warning: unclassified R2 actions ignored: " + ", ".join(bad), file=sys.stderr) - out.update(parse_r2(r2_rows)) - return out - - -def discover_namespaces(client, script, start, end): - lo, hi = minute_window(start, end) - groups = client.run(q_do_groups(script, lo, hi - timedelta(seconds=1))) - return {g["dimensions"]["namespaceId"] for g in groups["durableObjectsInvocationsAdaptiveGroups"]} - - -def read_all(client, args, phases, window): - namespaces = discover_namespaces(client, args.script, *window) - minutes = periodic_ranges(phases, getattr(args, 'steady', ())) - return {name: read_phase(client, args.script, args.bucket, namespaces, s, e, minutes[name]) - for name, (s, e) in phases.items()} - - -# ---------- orchestration ---------- - -def load_phases(args): - """Return ({name: (start, end)}, (window_start, window_end), script).""" - if args.manifest: - m = json.load(open(args.manifest)) - phases = {p["name"]: (parse_time(p["start"]), parse_time(p["end"])) for p in m["phases"]} - args.steady = {p["name"] for p in m["phases"] if p.get("steady")} - window = (parse_time(m["start"]), parse_time(m["end"])) - return phases, window, m.get("script") - if not (args.time_from and args.time_to): - sys.exit("usage: give --manifest FILE, or both --from and --to (or --check)") - window = (parse_time(args.time_from), parse_time(args.time_to)) - return {"total": window}, window, None - - -def wait_until(target): - while True: - left = (target - datetime.now(timezone.utc)).total_seconds() - if left <= 0: - return - print("waiting %ds for analytics to settle (until %s)" % (left, iso(target)), flush=True) - time.sleep(min(left, 60)) - - -def read_stable(client, args, phases, window): - """Analytics lag the events, so re-read each minute until two reads agree (max 20 minutes).""" - deadline = time.monotonic() + 20 * 60 - prev = read_all(client, args, phases, window) - while time.monotonic() < deadline: - time.sleep(60) - cur = read_all(client, args, phases, window) - if cur == prev: - return cur - print("numbers still moving, re-reading in 60s", flush=True) - prev = cur - print("warning: numbers never settled within 20 minutes", file=sys.stderr) - return prev - - -def build_result(args, phases_read, window): - now = datetime.now(timezone.utc) - return {"script": args.script, "bucket": args.bucket, "read_at": iso(now), - "lag_seconds": max(0, int((now - window[1]).total_seconds())), - "phases": phases_read, "total": sum_phases(phases_read)} - - -def check(client, args): - end = datetime.now(timezone.utc) - client.run(q_workers(args.script, end - timedelta(hours=1), end)) - print("OK: analytics API reachable and the token has the Analytics scope") - - -def parse_args(argv): - p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) - p.add_argument("--manifest") - p.add_argument("--from", dest="time_from") - p.add_argument("--to", dest="time_to") - p.add_argument("--script", default=None) - p.add_argument("--bucket", default=DEFAULT_BUCKET) - p.add_argument("--account", default=DEFAULT_ACCOUNT) - p.add_argument("--out") - p.add_argument("--settle-seconds", type=int, default=360) - p.add_argument("--stable", action=argparse.BooleanOptionalAction, default=True) - p.add_argument("--check", action="store_true") - return p.parse_args(argv) - - -def run(args): - client = Client(args.account) - if args.check: - return check(client, args) - phases, window, manifest_script = load_phases(args) - args.script = args.script or manifest_script or DEFAULT_SCRIPT - if args.settle_seconds > 0: # zero means the caller wants to read right now, even a future-ending window - wait_until(window[1] + timedelta(seconds=args.settle_seconds)) - read = read_stable if args.stable else read_all - result = build_result(args, read(client, args, phases, window), window) - text = json.dumps(result, indent=2) - if args.out: - open(args.out, "w").write(text + "\n") - print(text) - - -def main(argv=None): - args = parse_args(argv) - args.script = args.script or (None if args.manifest else DEFAULT_SCRIPT) - try: - run(args) - except ScopeError: - print("error: " + SCOPE_FIX, file=sys.stderr) - return 3 - except (AuthExpired, urllib.error.URLError, RuntimeError) as err: - print("error: %s" % (err or "token rejected (401) even after refresh"), file=sys.stderr) - return 4 - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/relay/hosted/cost/lines.py b/relay/hosted/cost/lines.py deleted file mode 100644 index 010fa1952a..0000000000 --- a/relay/hosted/cost/lines.py +++ /dev/null @@ -1,75 +0,0 @@ -#!/usr/bin/env python3 -"""Bill per Cloudflare line item at 1k, 10k and 100k users (25% active), for any number of billed runs. - -Reuses bill.py's month arithmetic (one source of truth for the shape and the prices) and adds the two lines -bill.py leaves out: R2 storage, priced from the frame puts the client sends, and WebSocket hibernation, -shown as the share of the open-home seconds the Durable Object was billed active. - - python3 lines.py --column TITLE MANIFEST BILLED [--column ...] -""" -import argparse - -import bill -from model import PRICE, over - -SCALES = [(1_000, 0.25), (10_000, 0.25), (100_000, 0.25)] -FRAME_BYTES = 43_200 # model.Shape.flush_bytes: the measured mean frame -LINES = ["Worker requests", "Worker CPU ms", "DO requests", "DO GB-s", "DO rows read", "DO rows written", - "R2 Class A", "R2 Class B"] - - -def frame_puts(manifest, shape): - """Frame puts per user-month, from each move phase's own count of puts per move.""" - phases = {p["name"]: p for p in manifest["phases"]} - per_move = lambda n: phases[n].get("client_requests", {}).get("POST /v1/store/frames", 0) / phases[n]["count"] - return shape["warm_per_month"] * per_move("warm_moves") + shape["cold_per_month"] * per_move("cold_moves") - - -def storage_dollars(puts, users): - """One month of retained frames; the first 10 GB are free.""" - gb = puts * FRAME_BYTES * users / 1e9 - return over(gb * 1e6, PRICE["r2_gb_free"] * 1e6, PRICE["r2_gb_month"]), gb - - -def awake_share(manifest, billed, shape): - """Billed active seconds over the open-home seconds in the steady phase: what hibernation leaves awake.""" - idle = next(p for p in manifest["phases"] if p["name"] == "idle_hold") - return billed["phases"]["idle_hold"]["do_active_s"] / idle["seconds"] - - -def column(c, shape): - month = bill.month_usage(c["manifest"], c["billed"], shape) - puts = frame_puts(c["manifest"], shape) - rows = {} - for n, active in SCALES: - m = {name: d for name, _, _, d in bill.meters(bill.scale(month, n * active), 1, 1)} - storage, gb = storage_dollars(puts, n * active) - rows[n] = {**m, "R2 storage": storage, "_base": m["Workers base"], "_gb": gb} - return month, rows, awake_share(c["manifest"], c["billed"], shape) - - -def main(argv=None): - ap = argparse.ArgumentParser() - ap.add_argument("--column", nargs=3, action="append", required=True, metavar=("TITLE", "MANIFEST", "BILLED")) - a = ap.parse_args(argv) - cols = [dict(title=t, manifest=bill.load(m), billed=bill.load(b)) for t, m, b in a.column] - shape = bill.shape_for(cols[0]["manifest"], {}) - out = [column(c, shape) for c in cols] - names = LINES + ["R2 storage"] - print("Usage per user-month") - for label, k in bill.USAGE_ROWS: - print(f"| {label} | " + " | ".join(bill.num(m[k]) for m, _, _ in out) + " |") - print("| DO active s (billed) | " + " | ".join(bill.num(m["do_active_s"]) for m, _, _ in out) + " |") - print("| awake share of open-home time | " + " | ".join(f"{s * 100:.2f}%" for _, _, s in out) + " |") - for n, _ in SCALES: - print(f"\nDollars at {n:,} users, 25% active") - print("| Line | " + " | ".join(c["title"] for c in cols) + " |") - for name in ["Workers base"] + names: - key = "_base" if name == "Workers base" else name - print(f"| {name} | " + " | ".join(f"${r[n][key]:,.2f}" for _, r, _ in out) + " |") - print("| **Total** | " + " | ".join(f"**${sum(r[n][k] for k in ['_base'] + names):,.2f}**" for _, r, _ in out) + " |") - print("| frames stored GB | " + " | ".join(f"{r[n]['_gb']:.2f}" for _, r, _ in out) + " |") - - -if __name__ == "__main__": - main() diff --git a/relay/hosted/cost/model.py b/relay/hosted/cost/model.py deleted file mode 100644 index 783e7e8527..0000000000 --- a/relay/hosted/cost/model.py +++ /dev/null @@ -1,128 +0,0 @@ -#!/usr/bin/env python3 -"""Monthly cost of the hosted directory, three ways, from counts measured with the real client. - -Measured inputs (real client, local `wrangler dev` then staging Workers; see STAGE-1H-DECISION.md section 8): - s1 55 s 6 sealed turns 6 flushes 21 requests R2 A 20 B 14 176,611 B up - s2 110 s 7 sealed turns 2 flushes 17 requests R2 A 16 B 15 141,824 B up - s3 takeover: 90 object GETs, 122 R2 B, 61,364 B down, 20 heartbeats over 3 min -Prices (Cloudflare docs fetched 2026-09-29): see PRICE below. -Everything else is a stated assumption and a parameter. -""" -from dataclasses import dataclass, replace - -PRICE = dict( - workers_base=5.0, req_free=10e6, req_per_m=0.30, cpu_free_ms=30e6, cpu_per_m_ms=0.02, - do_req_free=1e6, do_req_per_m=0.15, do_gbs_free=400e3, do_gbs_per_m=12.50, - do_rows_free=50e6, do_rows_per_m=1.0, r2_a_free=1e6, r2_a_per_m=4.50, r2_b_free=10e6, r2_b_per_m=0.36, - r2_gb_free=10, r2_gb_month=0.015, d1_rows_free=50e6, d1_rows_per_m=1.0, -) -over = lambda used, free, per_m: max(0.0, used - free) / 1e6 * per_m - - -@dataclass(frozen=True) -class Shape: - """One held session (30 minutes) as the client produces it.""" - heartbeats: float = 180 # client today: every 10 s. Renew-on-publish lease, TTL 90 s: 30 (a keepalive each 30 s, and only while nothing was published) - flushes: float = 60 # one frame put plus one publish each; 5 s flush interval, one per 30 s on average - fixed_writes: float = 3 # create, device record, release (measured: s1) - fixed_reads: float = 4 # the reads that come with them (measured: s1) - takeover: float = 0.2 # share of sessions that end in a takeover - take_gets: float = 96 # object GETs per takeover (billed: 193 requests in the dumb run less its polls and beats) - take_dir: float = 8 # directory requests per takeover (measured: s3) - take_b: float = 33 # extra R2 reads per takeover beyond its GETs (measured: s3 122 - 90 + snapshot) - flush_bytes: float = 43_200 # measured mean frame: 388,592 B over 9 flushes - do_ms: float = 40 # billed active ms per Durable Object request (staging: 14.3 s over 379 requests, 8.0 s over 194) - cpu_ms: float = 1.5 # billed CPU per Worker request (staging: 325 ms over 192, 378 over 379, 138 over 193) - sessions: float = 8 # per identity per month - home_hours: float = 0 # hours a home screen is open per identity per month; it polls the directory list - polls_per_hour: float = 450 # measured: one list per 8 s with this branch's client (a newer build polls every 3 s: 1,140) - - -def polls(s, n_identities): # list requests a month from open home screens - return s.home_hours * s.polls_per_hour * n_identities - - -def requests(s): # Worker invocations per session - return s.heartbeats + 2 * s.flushes + s.fixed_writes + s.fixed_reads / 2 + s.takeover * (s.take_gets + s.take_dir) - - -def dir_writes(s): - return s.heartbeats + s.flushes + s.fixed_writes - - -def storage(s, n_sessions, months=12): - gb = n_sessions * s.flush_bytes * s.flushes * months / 1e9 - return over(gb * 1e6, PRICE['r2_gb_free'] * 1e6, PRICE['r2_gb_month']) # (gb-free) * $/GB - - -def common(s, n_sessions, n_polls=0): - """Worker requests, CPU, frames in R2, storage: the same for every option.""" - req = requests(s) * n_sessions + n_polls - frames_a = s.flushes * n_sessions + s.takeover * n_sessions # frames plus one snapshot write per takeover - gets_b = s.takeover * s.take_gets * n_sessions - return dict( - workers=PRICE['workers_base'] + over(req, PRICE['req_free'], PRICE['req_per_m']) - + over(req * s.cpu_ms, PRICE['cpu_free_ms'], PRICE['cpu_per_m_ms']), - r2_a=frames_a, r2_b=gets_b, storage=storage(s, n_sessions), - ) - - -def option_a(s, n, p=0): # Durable Object per identity holds the directory (SQLite) and sees every put - c = common(s, n, p) - reqs = p + (dir_writes(s) + s.flushes + s.fixed_reads / 2) * n # dir ops plus puts route to the object - gbs = (reqs / n * s.do_ms / 1000 * 0.128) * n - rows = 2 * dir_writes(s) * n - c['do'] = over(reqs, PRICE['do_req_free'], PRICE['do_req_per_m']) + over(gbs, PRICE['do_gbs_free'], PRICE['do_gbs_per_m']) \ - + over(rows, PRICE['do_rows_free'], PRICE['do_rows_per_m']) - return c - - -def option_b(s, n, p=0): # D1 shard holds the directory; frames still go through the stateless Worker - c = common(s, n, p) - c['d1'] = over(2 * dir_writes(s) * n, PRICE['d1_rows_free'], PRICE['d1_rows_per_m']) - return c - - -def option_c(s, n, p=0): # R2 only: each directory write is one conditional put, each read one get - c = common(s, n, p) - c['r2_a'] += dir_writes(s) * n + s.takeover * n * 2 + p # each list is a Class A ListObjects # + acquire writes at takeover - c['r2_b'] += (s.heartbeats + s.flushes + s.fixed_reads) * n + s.takeover * n * (s.take_dir + s.take_b) - return c - - -def total(c): - return c['workers'] + c.get('do', 0) + c.get('d1', 0) + c['storage'] \ - + over(c['r2_a'], PRICE['r2_a_free'], PRICE['r2_a_per_m']) + over(c['r2_b'], PRICE['r2_b_free'], PRICE['r2_b_per_m']) - - -OPTIONS = {'a Durable Object': option_a, 'b D1': option_b, 'c R2 only': option_c} -CLIENTS = {'client today (beat 10 s)': {}, 'lease renews on publish': dict(heartbeats=30)} - - -def table(shape, label): - print(f'\n{label}') - print(f'{"":26}{"identities":>11}{"active":>8}' + ''.join(f'{k:>20}' for k in OPTIONS)) - for cname, tweak in CLIENTS.items(): - for n in (10_000, 100_000): - for active in (1.0, 0.25): - s = replace(shape, **tweak) - sessions = n * active * s.sessions - row = [total(f(s, sessions, polls(s, n * active))) for f in OPTIONS.values()] - print(f'{cname:26}{n:>11,}{active:>8.0%}' + ''.join(f'{v:>20,.0f}' for v in row)) - - -if __name__ == '__main__': - table(Shape(takeover=1.0), 'UPPER BOUND: every identity active, 8 sessions of 30 min, a takeover in every session (demo shape), a flush every 30 s') - table(Shape(), 'REALISTIC SHAPE: same activity, takeover in 20% of sessions') - table(Shape(flushes=360, takeover=1.0), 'STRESS: a flush every 5 s for all 30 minutes (never idle), takeover in every session') - table(Shape(flushes=12), 'QUIET: a flush every 2.5 minutes') - - -def breakeven_resident_s(shape, n_identities, active=1.0): - """Seconds of billed Durable Object residency per session at which option a costs as much as option c.""" - n = n_identities * active * shape.sessions - a, c = option_a(shape, n), option_c(shape, n) - gap = total(c) - total(a) - modelled = (requests(shape) * shape.do_ms / 1000 * 0.128) * n - need = max(PRICE['do_gbs_free'], modelled) + gap / PRICE['do_gbs_per_m'] * 1e6 - return (need - modelled) / n / 0.128 diff --git a/relay/hosted/cost/quiet.py b/relay/hosted/cost/quiet.py deleted file mode 100644 index 752fc57c5b..0000000000 --- a/relay/hosted/cost/quiet.py +++ /dev/null @@ -1,67 +0,0 @@ -#!/usr/bin/env python3 -"""Re-read a steady phase from only its quiet minutes, when another session shared the staging Worker. - -The billed analytics are per Worker, so traffic from anyone else lands in the phase. A steady phase -(the idle hold) is one client doing almost nothing, so a minute with more than --max-requests Worker -requests cannot be that client alone. This keeps only the minutes at or under the threshold (a minute -with no data is quiet) and writes a manifest and a billed file whose phase is exactly those minutes, so -the per-second rate that bill.py extrapolates is taken over clean time only. The other phases are copied. - - python3 quiet.py --manifest run.json --billed billed.json --out-prefix PATH [--max-requests 3] -""" -import argparse -import json -from datetime import timedelta - -import billed - -MINUTE = timedelta(minutes=1) - - -def worker_minutes(client, script, lo, hi): - q = ('workersInvocationsAdaptive(limit:1000, filter:{scriptName:"%s", datetime_geq:"%s", datetime_leq:"%s"}) ' - '{ dimensions { datetimeMinute } sum { requests } }' % (script, billed.iso(lo), billed.iso(hi - timedelta(seconds=1)))) - rows = client.run(q)["workersInvocationsAdaptive"] - return {billed.parse_time(r["dimensions"]["datetimeMinute"]): r["sum"]["requests"] for r in rows} - - -def quiet_minutes(counts, lo, hi, limit): - out, m = [], lo - while m < hi: - if counts.get(m, 0) <= limit: - out.append(m) - m += MINUTE - return out - - -def main(argv=None): - ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) - ap.add_argument("--manifest", required=True) - ap.add_argument("--billed", required=True) - ap.add_argument("--out-prefix", required=True) - ap.add_argument("--max-requests", type=int, default=3) - ap.add_argument("--bucket", default=billed.DEFAULT_BUCKET) - ap.add_argument("--account", default=billed.DEFAULT_ACCOUNT) - a = ap.parse_args(argv) - manifest, result = json.load(open(a.manifest)), json.load(open(a.billed)) - client = billed.Client(a.account) - phase = next(p for p in manifest["phases"] if p.get("steady")) - lo, hi = billed.minute_window(billed.parse_time(phase["start"]), billed.parse_time(phase["end"])) - keep = quiet_minutes(worker_minutes(client, manifest["script"] if "script" in manifest else result["script"], lo, hi), lo, hi, a.max_requests) - script = result["script"] - namespaces = billed.discover_namespaces(client, script, lo, hi) - reads = [billed.read_phase(client, script, a.bucket, namespaces, m, m + MINUTE, (m, m)) for m in keep] - summed = billed.sum_phases(dict(enumerate(reads))) - result["phases"][phase["name"]] = summed - result["total"] = billed.sum_phases(result["phases"]) - phase["seconds"] = 60.0 * len(keep) - phase["note"] = (f"read from the {len(keep)} quiet minutes of {int((hi - lo) / MINUTE)} (at most {a.max_requests} Worker requests " - "a minute) because another session was loading the staging Worker in the rest") - json.dump(manifest, open(a.out_prefix + ".run.json", "w"), indent=2) - json.dump(result, open(a.out_prefix + ".billed.json", "w"), indent=2) - print(f"kept {len(keep)} minutes: " + " ".join(m.strftime("%H:%M") for m in keep)) - print(json.dumps(summed)) - - -if __name__ == "__main__": - main() diff --git a/relay/hosted/cost/test_bill.py b/relay/hosted/cost/test_bill.py deleted file mode 100644 index 23f9618da9..0000000000 --- a/relay/hosted/cost/test_bill.py +++ /dev/null @@ -1,77 +0,0 @@ -import unittest - -import bill -from model import PRICE - -MANIFEST = {"url": "u", "script": "s", "start": "2026-01-01T00:00:00Z", "end": "2026-01-01T00:30:00Z", - "phases": [{"name": "idle_hold", "seconds": 100}, {"name": "warm_moves", "count": 10}, - {"name": "cold_moves", "count": 2}]} - - -def vec(**kw): - return {k: kw.get(k, 0) for k in bill.KEYS} - - -def billed(scale=1): - return {"script": "heavy", "phases": { - "idle_hold": vec(do_requests=50 * scale, do_gb_s=2 * scale, worker_requests=60), - "warm_moves": vec(do_requests=100 * scale, do_gb_s=1 * scale, r2_class_a=20), - "cold_moves": vec(do_requests=40 * scale, do_gb_s=4 * scale, r2_class_b=30)}} - - -class BillTest(unittest.TestCase): - def test_extrapolation(self): - u = bill.month_usage(MANIFEST, billed(), bill.SHAPE) - bg = 8 * 3600 * 22 / 100 # 6336 times the 100 s phase - self.assertAlmostEqual(u["do_requests"], 50 * bg + 100 / 10 * 40 + 40 / 2 * 4) - self.assertAlmostEqual(u["do_gb_s"], 2 * bg + 1 / 10 * 40 + 4 / 2 * 4) - self.assertAlmostEqual(u["worker_requests"], 60 * bg) - self.assertAlmostEqual(u["r2_class_a"], 20 / 10 * 40) - self.assertAlmostEqual(u["r2_class_b"], 30 / 2 * 4) - - def test_shape_override(self): - shape = bill.shape_for({"shape": {"warm_per_month": 1}}, {"cold_per_month": 0, "days_per_month": None}) - self.assertEqual((shape["warm_per_month"], shape["cold_per_month"], shape["days_per_month"]), (1, 0, 22)) - - def test_charge(self): - u = vec(do_requests=2e6, r2_class_a=3e6) - self.assertAlmostEqual(bill.charge(u, 0, 0), 2 * PRICE["do_req_per_m"] + 3 * PRICE["r2_a_per_m"]) - self.assertAlmostEqual(bill.charge(u, 1, 1), PRICE["workers_base"] + 1 * PRICE["do_req_per_m"] - + 2 * PRICE["r2_a_per_m"]) - - def test_rows_priced_apart(self): - self.assertAlmostEqual(bill.charge(vec(do_rows_read=1e6, do_rows_written=1e6), 0, 0), bill.ROWS_READ["per_m"] + bill.ROWS_WRITTEN["per_m"]) - - def test_render_one_column(self): - out = bill.render([dict(title="first", manifest=MANIFEST, billed=billed())], bill.SHAPE) - self.assertTrue(out.startswith(bill.HEADING + "\n")) - self.assertIn("| Usage | first |", out) - self.assertIn("100,000 users, 100% active", out) - self.assertIn("**Total**", out) - - def test_render_many_columns_and_ratios(self): - cols = [dict(title=t, manifest=MANIFEST, billed=billed(k)) for t, k in (("one", 1), ("two", 0.5), ("three", 0.25))] - out = bill.render(cols, bill.SHAPE) - self.assertIn("| Usage | one | two | three |", out) - self.assertIn("Ratio two / one per user-month: DO GB-s 0.50, DO requests 0.50", out) - self.assertIn("Ratio three / one per user-month: DO GB-s 0.25", out) - - def test_class_a_is_at_least_the_client_frame_puts(self): - phase = {"client_requests": {"POST /v1/store/frames": 12}} - self.assertEqual(bill.class_a_floor({"r2_class_a": 0}, phase)["r2_class_a"], 12) - self.assertEqual(bill.class_a_floor({"r2_class_a": 30}, phase)["r2_class_a"], 30) - - def test_splice_replace_and_append(self): - sec = bill.HEADING + "\n\nnew\n" - doc = "# T\n\n## 8. x\n\nbody\n\n## 9. Billed heavy-user run\n\nold\n\n## 10. y\n\nz\n" - self.assertEqual(bill.splice(doc, sec), - "# T\n\n## 8. x\n\nbody\n\n## 9. Billed heavy-user run\n\nnew\n\n## 10. y\n\nz\n") - last = "# T\n\n## 9. Billed heavy-user run\n\nold\n" - self.assertEqual(bill.splice(last, sec), "# T\n\n" + sec) - self.assertEqual(bill.splice("# T\n\nbody\n", sec), "# T\n\nbody\n\n" + sec) - again = bill.splice(bill.splice("# T\n", sec), sec) - self.assertEqual(again, bill.splice("# T\n", sec)) - - -if __name__ == "__main__": - unittest.main() diff --git a/relay/hosted/cost/test_billed.py b/relay/hosted/cost/test_billed.py deleted file mode 100644 index e2147eef89..0000000000 --- a/relay/hosted/cost/test_billed.py +++ /dev/null @@ -1,51 +0,0 @@ -import unittest -from datetime import datetime, timezone - -import billed - - -def r2(action, status, n): - return {"dimensions": {"actionType": action, "actionStatus": status}, "sum": {"requests": n}} - - -class BilledTest(unittest.TestCase): - def test_classify(self): - self.assertEqual(billed.classify("PutObject", "success"), "A") - self.assertEqual(billed.classify("ListObjects", "success"), "A") - self.assertEqual(billed.classify("GetObject", "notFound"), "B") - self.assertEqual(billed.classify("HeadBucket", "success"), "B") - self.assertIsNone(billed.classify("DeleteObject", "success")) - self.assertIsNone(billed.classify("AbortMultipartUpload", "success")) - self.assertIsNone(billed.classify("PutObject", "serverError")) - self.assertIsNone(billed.classify("SomethingNew", "success")) - - def test_parse_r2(self): - rows = [r2("PutObject", "success", 27), r2("GetObject", "success", 130), - r2("GetObject", "notFound", 3), r2("DeleteObject", "success", 9)] - self.assertEqual(billed.parse_r2(rows), {"r2_class_a": 27, "r2_class_b": 133}) - self.assertEqual(billed.unknown_actions(rows + [r2("Zap", "success", 1)]), ["Zap"]) - - def test_parse_workers_and_do(self): - w = billed.parse_workers([{"sum": {"requests": 1247, "cpuTimeUs": 982911}}]) - self.assertEqual(w, {"worker_requests": 1247, "worker_cpu_ms": 982.911}) - rows = [{"sum": {"activeTime": 14_300_000, "duration": 1.83, "rowsRead": 5, "rowsWritten": 2}}, - {"sum": {"activeTime": 0, "duration": 0.0, "rowsRead": 1, "rowsWritten": 0}}] - d = billed.parse_do_periodic(rows) - self.assertEqual((d["do_active_s"], d["do_gb_s"], d["do_rows_read"]), (14.3, 1.83, 6)) - self.assertEqual(billed.parse_do_invocations([{"sum": {"requests": 4}}, {"sum": {"requests": 6}}]), 10) - - def test_minute_window(self): - s = datetime(2026, 9, 30, 10, 0, 40, tzinfo=timezone.utc) - e = datetime(2026, 9, 30, 10, 5, 0, tzinfo=timezone.utc) - lo, hi = billed.minute_window(s, e) - self.assertEqual((billed.iso(lo), billed.iso(hi)), ("2026-09-30T10:00:00Z", "2026-09-30T10:05:00Z")) - _, hi2 = billed.minute_window(s, e.replace(second=1)) - self.assertEqual(billed.iso(hi2), "2026-09-30T10:06:00Z") - - def test_sum_phases(self): - a = dict.fromkeys(billed.METRICS, 1) - self.assertEqual(billed.sum_phases({"x": a, "y": a})["r2_class_a"], 2) - - -if __name__ == "__main__": - unittest.main() diff --git a/relay/hosted/production.toml b/relay/hosted/production.toml deleted file mode 100644 index 31b43965c6..0000000000 --- a/relay/hosted/production.toml +++ /dev/null @@ -1,75 +0,0 @@ -# Production deployment, served at https://codeaf.agentfield.ai/fabric and the pairing link page -# at https://codeaf.agentfield.ai/p/, and nowhere else. -# -# The hostname already answers other things, so the Worker owns only two paths: /fabric*, the wire, -# and /p/*, the page a pairing link opens in a browser. That page lives on the bare host, outside -# BASE_PATH, so it needs its own route. The routes below leave every other path on -# codeaf.agentfield.ai exactly as it is, and BASE_PATH makes the Worker serve its wire beneath -# /fabric and answer 404 for every other path that reaches it. -# -# Nothing here deploys by itself. Create what it names once, from a machine logged in to the -# Cloudflare account that holds the agentfield.ai zone: -# -# npx wrangler r2 bucket create caf-fabric-frames -# npx wrangler deploy -c production.toml -# -# The Durable Object namespaces are created by the first deploy, from the migrations below, and are -# named after this Worker (caf-fabric-IdentityDO and so on), so they share nothing with staging. -# -# What staging sets and production must not: TRUST_PROXY (production's one truth for a caller's -# network is CF-Connecting-IP) and a CAF_LIMITS override (the new-identity cap per address and the -# pairing time to live stay at the contract defaults). `LEASE_POLICY` is the amended law the clients -# implement. Observability stays off: the relay keeps no request log, as its README states. -name = "caf-fabric" -main = "src/worker.js" -compatibility_date = "2026-09-01" -workers_dev = false -preview_urls = false - -routes = [ - { pattern = "codeaf.agentfield.ai/fabric*", zone_name = "agentfield.ai" }, - { pattern = "codeaf.agentfield.ai/p/*", zone_name = "agentfield.ai" }, -] - -[observability] -enabled = false - -[[r2_buckets]] -binding = "FRAMES" -bucket_name = "caf-fabric-frames" - -[[durable_objects.bindings]] -name = "IDENTITY" -class_name = "IdentityDO" - -[[durable_objects.bindings]] -name = "PAIR_GATE" -class_name = "PairGate" - -[[durable_objects.bindings]] -name = "NEWCOMERS" -class_name = "NewcomerGate" - -[[durable_objects.bindings]] -name = "MAILBOX" -class_name = "Mailbox" - -[[durable_objects.bindings]] -name = "LINK_GATE" -class_name = "LinkGate" - -[[durable_objects.bindings]] -name = "LINK_REQUEST" -class_name = "LinkRequest" - -[[migrations]] -tag = "v1" -new_sqlite_classes = ["IdentityDO", "PairGate", "Mailbox", "NewcomerGate"] - -[[migrations]] -tag = "v2" -new_sqlite_classes = ["LinkGate", "LinkRequest"] - -[vars] -LEASE_POLICY = "amended" -BASE_PATH = "/fabric" diff --git a/relay/hosted/src/limits.js b/relay/hosted/src/limits.js index 85e36fa5b2..d38b1d3a40 100644 --- a/relay/hosted/src/limits.js +++ b/relay/hosted/src/limits.js @@ -1,6 +1,6 @@ // Every number the relay enforces, in one table, and the two small mechanisms that enforce them. -// Free forever needs ceilings: the money-bearing ones (bytes, objects, frames a day) are kept in -// the identity's Durable Object storage and survive eviction; the per-minute rates are kept in +// Stored quotas (bytes, objects, frames a day) survive eviction in the identity's Durable Object +// storage; the per-minute rates are kept in // memory, because an eviction can only forgive a burst, never let a stored quota be exceeded. import { full, rateLimited } from './wire.js'; @@ -8,7 +8,7 @@ const GiB = 1 << 30; const MiB = 1 << 20; /** - * The numbers of presence (contract 21.12.3), each said once. A client declares the beat it pings at; a socket that + * Presence timing defaults, each said once. A client declares the beat it pings at; a socket that * declares none is a pre-presence client, which was told to ping every 30 s. The window is how many beats of silence * the relay allows before the socket stops counting as proof that its device is alive. */ @@ -19,27 +19,27 @@ export const windowMs = (beat) => beat * 1000 * PRESENCE.windowFactor; export const DEFAULTS = { // One identity. - storeBytes: 5 * GiB, // stored bytes; the cost ceiling + storeBytes: 5 * GiB, // Maximum stored bytes per identity. storeObjects: 5_000_000, // the index is rows in the identity's SQLite, so this is a backstop; the byte cap rules - framesPerDay: 5_000, // R2 writes, the dearest operation + framesPerDay: 5_000, // Maximum new frames stored per identity per day. requestsPerMinute: 1_200, // all devices of one identity together requestsPerMinutePerDevice: 600, statsFlushMs: 5_000, // how long a count may wait in memory before it is written (see stats.js) concurrentPuts: 2, // a 16 MiB frame costs the isolate 2 to 3 times its size while it arrives - // Rotation (contract 20.3): how long a retired identity stays readable before it is deleted. + // Rotation: how long a retired identity stays readable before it is deleted. // The minimum and the page size are the test hooks, like the self-hosted relay's --min-grace. minGraceMs: 3_600_000, maxGraceMs: 30 * 86_400_000, defaultGraceMs: 7 * 86_400_000, - maxWatchers: 1_000, // directory watch sockets one identity may hold (contract 21.8); the Go relay's directory.MaxWatchers - maxHolds: 16, // cells one watch socket may vouch for (contract 21.11.1); the Go relay's directory.MaxHolds + maxWatchers: 1_000, // directory watch sockets one identity may hold; the Go relay's directory.MaxWatchers + maxHolds: 16, // cells one watch socket may vouch for; the Go relay's directory.MaxHolds sweepPageSize: 1_000, // R2 objects deleted per alarm turn (R2's own limit per list and per delete) - // The whole relay, by caller IP: how many identities one address may bring in (contract 6: no accounts). + // The whole relay, by caller IP: how many identities one address may bring in. newIdentitiesPerIpPerDay: 200, - // Deployment-wide brakes remain effective when a caller spreads across networks. - globalNewIdentitiesPerDay: 10_000, - globalStoreBytes: 1024 * GiB, - // The pairing mailbox (contract 18.4), for callers that have no identity yet. + // Deployment-wide brakes are opt-in through CAF_LIMITS; zero closes the corresponding door. + globalNewIdentitiesPerDay: Infinity, + globalStoreBytes: Infinity, + // The pairing mailbox, for callers that have no identity yet. pairTtlMs: 10 * 60_000, pairMaxMsg: 4 << 10, pairMaxMsgsPerSide: 4, @@ -49,7 +49,7 @@ export const DEFAULTS = { pairMaxBoxes: 2_000, // at 32 KiB each this is also the 64 MiB ceiling in all // How long an ended mailbox's nameplate waits before it can be drawn again; null means pairTtlMs (see pair/plates.js). plateQuarantineMs: null, - // Link requests (docs/ux-pairing-contract.md 3.6), for a new device that has no identity yet. The + // Link requests, for a new device that has no identity yet. The // time to live is pairTtlMs, shared with the mailbox. linkCreatePerHour: 10, linkPendingPerIp: 3, @@ -98,7 +98,7 @@ const DAY_MS = 86_400_000; /** * Quota is one identity's stored totals, kept in its Durable Object's SQLite: bytes and objects * held, and the frames stored today. admit refuses a frame that would pass a ceiling, before - * it costs anything; record counts a frame once R2 has confirmed it is new. + * it is written; record counts a frame once R2 has confirmed it is new. */ export class Quota { constructor(sql, limits) { diff --git a/relay/hosted/staging.toml b/relay/hosted/staging.toml deleted file mode 100644 index 8886943fb2..0000000000 --- a/relay/hosted/staging.toml +++ /dev/null @@ -1,51 +0,0 @@ -# Staging deployment, on workers.dev only: no route, no domain, no observability. -# BASE_PATH is empty here: staging owns its whole workers.dev hostname. -# Run from a machine logged in to the account: `wrangler r2 bucket create caf-relay-staging-frames` -# once, then `wrangler deploy -c staging.toml`. It owns only what is named here. -name = "caf-relay-staging" -main = "src/worker.js" -compatibility_date = "2026-09-01" - -[[r2_buckets]] -binding = "FRAMES" -bucket_name = "caf-relay-staging-frames" - -[[durable_objects.bindings]] -name = "IDENTITY" -class_name = "IdentityDO" - -[[durable_objects.bindings]] -name = "PAIR_GATE" -class_name = "PairGate" - -[[durable_objects.bindings]] -name = "NEWCOMERS" -class_name = "NewcomerGate" - -[[durable_objects.bindings]] -name = "MAILBOX" -class_name = "Mailbox" - -[[durable_objects.bindings]] -name = "LINK_GATE" -class_name = "LinkGate" - -[[durable_objects.bindings]] -name = "LINK_REQUEST" -class_name = "LinkRequest" - -[[migrations]] -tag = "v1" -new_sqlite_classes = ["IdentityDO", "PairGate", "Mailbox", "NewcomerGate"] - -[[migrations]] -tag = "v2" -new_sqlite_classes = ["LinkGate", "LinkRequest"] - -[vars] -LEASE_POLICY = "amended" -# The conformance suite makes one identity per case, all from one address, so the per-IP cap on new -# identities is lifted for staging only; production keeps the default (see src/limits.js). -# Staging trusts X-Forwarded-For so callers can name their own network; production does not. -TRUST_PROXY = "1" -CAF_LIMITS = '{"newIdentitiesPerIpPerDay":1000000,"pairTtlMs":4000,"minGraceMs":2000,"sweepPageSize":2,"requestsPerMinute":100000,"requestsPerMinutePerDevice":100000}' diff --git a/relay/hosted/test/deploy-local.sh b/relay/hosted/test/deploy-local.sh deleted file mode 100644 index f05c3d9a92..0000000000 --- a/relay/hosted/test/deploy-local.sh +++ /dev/null @@ -1,57 +0,0 @@ -#!/bin/sh -# C22 exercises the operator's smoke checks locally; C20 exercises the cost brakes on a fresh Worker. -set -eu -cd "$(dirname "$0")/.." -. ./test/proxy-env.sh -proof_root=$(mktemp -d "${TMPDIR:-/tmp}/codeaf-relay-proof.XXXXXX") -proof_pid= -cleanup() { - proof_status=$? - if [ -n "$proof_pid" ]; then kill "$proof_pid" 2>/dev/null || true; wait "$proof_pid" 2>/dev/null || true; fi - if [ "$proof_status" -eq 0 ]; then rm -rf "$proof_root"; - else echo "Local proof failed; evidence kept at $proof_root" >&2; fi -} -trap cleanup EXIT -trap 'exit 130' INT -trap 'exit 143' TERM -port=$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1",0)); print(s.getsockname()[1])') -start() { - config=$1; name=$2; shift 2 - sh test/local.sh node_modules/.bin/wrangler dev --local -c "$config" --name "$name" \ - --ip 127.0.0.1 --port "$port" --inspector-port 0 --persist-to "$proof_root/$name" "$@" >"$proof_root/$name.log" 2>&1 & - proof_pid=$! - for _ in $(seq 90); do - if curl -s -o /dev/null "http://127.0.0.1:$port/"; then return; fi - sleep 1 - done - cat "$proof_root/$name.log"; return 1 -} -stop() { - kill "$proof_pid" 2>/dev/null || true - wait "$proof_pid" 2>/dev/null || true - proof_pid= -} - -start production.toml "caf-production-local-$port" -page=$(curl -fsS "http://127.0.0.1:$port/p/k7m2q9xd") -printf '%s' "$page" | rg -q '' -status=$(curl -sS -o "$proof_root/missing.json" -w '%{http_code}' "http://127.0.0.1:$port/fabric/does-not-exist") -test "$status" = 404 -python3 -c 'import json,sys; assert json.load(open(sys.argv[1])) == {"err":"not_found"}' "$proof_root/missing.json" -echo 'C22: production routes, pairing page and 404 JSON pass locally' -page_retries=$(grep -Fc 'codeaf local proxy transport retry: Network connection lost.' "$proof_root/caf-production-local-$port.log" || true) -echo "C22: local proxy transport retries: $page_retries" -stop - -start staging.toml "caf-brakes-local-$port" --var 'CAF_LIMITS:{"globalNewIdentitiesPerDay":2,"globalStoreBytes":512,"minGraceMs":2000}' -RELAY="http://127.0.0.1:$port" node test/global.e2e.mjs >"$proof_root/global.log" 2>&1 || { cat "$proof_root/global.log"; exit 1; } -cat "$proof_root/global.log" -global_retries=$(grep -Fc 'codeaf local proxy transport retry: Network connection lost.' "$proof_root/caf-brakes-local-$port.log" || true) -echo "C20: local proxy transport retries: $global_retries" -stop - -# The canonical runner gives each suite its own report and covers the two-mailbox case too. -(cd ../.. && GOFLAGS='-buildvcs=false -p=2' GOMAXPROCS=4 \ - CONFORMANCE_TRACKS=worker CONFORMANCE_SERIAL=1 CONFORMANCE_WORK="$proof_root/conformance" \ - CONFORMANCE_WORKER_CONFIG="$PWD/relay/hosted/staging.toml" SKIP_E2E=1 bash scripts/relay-conformance.sh) -echo 'C22: staging conformance passes locally, including the capped second relay' diff --git a/relay/hosted/test/global.test.js b/relay/hosted/test/global.test.js index bf38d4cf91..dc9c5f77c0 100644 --- a/relay/hosted/test/global.test.js +++ b/relay/hosted/test/global.test.js @@ -1,7 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { GlobalBudget } from '../src/global.js'; -import { DEFAULTS } from '../src/limits.js'; +import { DEFAULTS, limitsOf } from '../src/limits.js'; import { Tenant } from '../src/tenant.js'; import { Flight } from '../src/flight.js'; import { policyOf } from '../src/rules.js'; @@ -13,6 +13,20 @@ const empty = { list: async () => ({ objects: [], truncated: false }) }; const refusal = (code, status) => (e) => e.code === code && e.status === status; const limits = (overrides) => ({ ...DEFAULTS, ...overrides }); +test('deployment defaults allow admission and storage above the former global limits', async () => { + const sql = memorySql(); + const budget = new GlobalBudget(sql, empty, limitsOf({})); + await budget.reserve('a', 'large', 2 ** 41); + for (let i = 0; i <= 10_000; i++) await budget.admit(`network-${i}`, 0); + await new GlobalBudget(sql, empty, limitsOf({})).admit('another', 0); +}); + +test('a zero storage override closes both admission and new writes', async () => { + const budget = new GlobalBudget(memorySql(), empty, limitsOf({ CAF_LIMITS: '{"globalStoreBytes":0}' })); + await assert.rejects(budget.admit('a', 0), refusal('relay_storage_full', 507)); + await assert.rejects(budget.reserve('a', 'f', 1), refusal('relay_storage_full', 507)); +}); + // C20 bounds successful new identities across networks and keeps the count after restart. test('global admission has one durable daily ceiling across networks', async () => { const sql = memorySql(), cap = limits({ globalNewIdentitiesPerDay: 2 }); diff --git a/relay/hosted/test/production.test.js b/relay/hosted/test/production.test.js deleted file mode 100644 index 228882d6f7..0000000000 --- a/relay/hosted/test/production.test.js +++ /dev/null @@ -1,33 +0,0 @@ -// The production configuration's front door: with its own BASE_PATH, the pairing page is served on -// the bare host and every other path outside /fabric is a 404, and its routes name exactly those two. -import assert from 'node:assert/strict'; -import { readFileSync } from 'node:fs'; -import { test } from 'node:test'; -import { dispatch } from '../src/front.js'; -import { respond } from '../src/wire.js'; - -const config = readFileSync(new URL('../production.toml', import.meta.url), 'utf8'); -const env = { BASE_PATH: /^BASE_PATH = "([^"]*)"/m.exec(config)[1] }; -const fetchAt = (path, method = 'GET') => respond(() => dispatch(new Request('https://codeaf.agentfield.ai' + path, { method }), env)); - -test('the production config routes the wire and the pairing page, and nothing else', () => { - const patterns = [...config.matchAll(/pattern = "([^"]+)"/g)].map((m) => m[1]); - assert.deepEqual(patterns, ['codeaf.agentfield.ai/fabric*', 'codeaf.agentfield.ai/p/*']); - assert.equal(env.BASE_PATH, '/fabric'); -}); - -test('a pairing link opens the page with the strict CSP', async () => { - const res = await fetchAt('/p/k7m2q9xd'); - assert.equal(res.status, 200); - assert.match(res.headers.get('content-type'), /^text\/html/); - const csp = res.headers.get('content-security-policy'); - assert.match(csp, /default-src 'none'/); - assert.doesNotMatch(csp, /unsafe|\*/); -}); - -test('every other path outside /fabric is a 404', async () => { - for (const path of ['/', '/anything', '/v1/dir/list', '/fabricx/v1/store/stats', '/p/', '/p/a/b', '/other/p/abc']) { - assert.equal((await fetchAt(path)).status, 404, path); - } - assert.equal((await fetchAt('/p/k7m2q9xd', 'POST')).status, 404); -}); diff --git a/relay/hosted/wrangler.toml b/relay/hosted/wrangler.toml index d2fcb7c89e..7ad35ea140 100644 --- a/relay/hosted/wrangler.toml +++ b/relay/hosted/wrangler.toml @@ -42,6 +42,5 @@ tag = "v2" new_sqlite_classes = ["LinkGate", "LinkRequest"] [vars] -# The lease law: "amended" is contract 8.7 of the stage 1H decision (90 s, a publish renews) and is -# what the client and the Go relay implement; "stage1" is the frozen 30 s law, kept for reference. +# "amended": 90-second leases, renewed by publish; "stage1": 30-second leases. LEASE_POLICY = "amended" diff --git a/relay/wrangler.toml b/relay/wrangler.toml index 2160d90b9b..a5696111d0 100644 --- a/relay/wrangler.toml +++ b/relay/wrangler.toml @@ -2,16 +2,13 @@ name = "codeaf-pool" main = "src/worker.js" compatibility_date = "2026-09-01" -# A route on the path prefix, not a custom domain: the Worker owns only -# codeaf.agentfield.ai/pool/* and the host can carry other services. -routes = [ - { pattern = "codeaf.agentfield.ai/pool/*", zone_name = "agentfield.ai" } -] +# Optional route for a domain you control. +# routes = [{ pattern = "relay.example.com/pool/*", zone_name = "example.com" }] [[kv_namespaces]] binding = "POOL" -# The runbook fills this id: `cd relay && wrangler kv namespace create POOL`. -id = "4b5cf70bed444ceb84a23ebf2be692e9" +# Create a namespace and replace this placeholder: `cd relay && wrangler kv namespace create POOL`. +id = "" # Observability as set in the dashboard, kept here so a redeploy does not # silently change it. The Worker itself logs nothing; what these persist are diff --git a/scripts/measure/hosted/cellread.py b/scripts/measure/hosted/cellread.py deleted file mode 100644 index 060b33dfc0..0000000000 --- a/scripts/measure/hosted/cellread.py +++ /dev/null @@ -1,42 +0,0 @@ -#!/usr/bin/env python3 -"""Read a cell's own on-disk record into JSON: `cellread.py `. - -Product signals only: turns.jsonl (sealed_at_ms, parent chain) and the receipt -each turn names (call started/ended in ms). Prints a list, oldest turn first: - {idx, id, sealed_ms, last_call_end_ms, seal_ms, calls, trigger, fence} -seal_ms = sealed_at_ms - the last call's end, the time the product took to seal -after the tool call finished (null when the turn has no call). -""" -import json -import os -import sys - - -def turns(cell): - with open(os.path.join(cell, "turns.jsonl")) as f: - return [json.loads(l) for l in f if l.strip()] - - -def receipt(cell, rid): - try: - with open(os.path.join(cell, "receipts", rid + ".json")) as f: - return json.load(f) - except FileNotFoundError: - return {} - - -def rows(cell): - out = [] - for i, t in enumerate(turns(cell)): - calls = receipt(cell, t.get("receipt", "")).get("calls", []) - end = max((c.get("ended", 0) for c in calls), default=None) - out.append({"idx": i, "id": t["id"], "sealed_ms": t["sealed_at_ms"], - "last_call_end_ms": end, - "seal_ms": t["sealed_at_ms"] - end if end else None, - "calls": len(calls), "trigger": t.get("trigger"), - "fence": t.get("fence")}) - return out - - -if __name__ == "__main__": - print(json.dumps(rows(sys.argv[1]))) diff --git a/scripts/measure/hosted/clockoff.py b/scripts/measure/hosted/clockoff.py deleted file mode 100644 index 5a2f1a4b63..0000000000 --- a/scripts/measure/hosted/clockoff.py +++ /dev/null @@ -1,40 +0,0 @@ -#!/usr/bin/env python3 -"""B's wall clock minus A's, in ms, by ping-pong over one open ssh channel. - - clockoff.py -> {"offset_ms": .., "rtt_ms": .., "samples": ..} - -A tiny loop on B answers each line with its clock at once, so the round trip is -the network and one ssh channel, not a process start. The reported offset is the -midpoint estimate from the lowest-round-trip sample; its error is at most half -that round trip, which is printed beside it. -""" -import json -import os -import subprocess -import sys -import time - -sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) -import rigenv # noqa: E402 - -REMOTE = 'import sys,time\nwhile sys.stdin.readline():\n print(int(time.time()*1000),flush=True)\n' - - -def main(control): - p = subprocess.Popen(["ssh", "-o", "BatchMode=yes", "-o", "ControlMaster=auto", "-o", f"ControlPath={control}", - "-o", "ControlPersist=20m", rigenv.get("CODEAF_SECOND_HOST"), "python3", "-u", "-c", "'" + REMOTE.replace("'", "") + "'"], - stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) - best = None - for _ in range(40): - t1 = time.time() * 1000 - p.stdin.write("x\n"); p.stdin.flush() - tb = int(p.stdout.readline()) - t2 = time.time() * 1000 - rtt = t2 - t1 - if best is None or rtt < best[0]: - best = (rtt, tb - (t1 + t2) / 2) - p.stdin.close(); p.wait() - print(json.dumps({"offset_ms": round(best[1]), "rtt_ms": round(best[0], 1), "samples": 40})) - - -main(sys.argv[1]) diff --git a/scripts/measure/hosted/durable.py b/scripts/measure/hosted/durable.py deleted file mode 100644 index 62b549a497..0000000000 --- a/scripts/measure/hosted/durable.py +++ /dev/null @@ -1,50 +0,0 @@ -#!/usr/bin/env python3 -"""Join a cell's sealed turns with the directory's head timeline. - - durable.py - -The watch file is s1probe's JSONL: one line per poll of the directory listing. -A turn becomes durable at the `durable_at` (directory clock, ms) the directory -gave the FIRST head that is this turn or a later one. Turns that never became -durable are counted, not dropped, because "nothing lost" is what item 7 checks. -seal_to_durable_ms = durable_at - sealed_at_ms. Both clocks are A's own here -(the relay container shares the host kernel clock); the watch lines also carry -the directory's `now` next to local ms so any skew is visible in the raw file. -""" -import json -import sys - - -def head_changes(watch, cell): - """(head, durable_at) each time the head moves, in poll order.""" - seen, out = None, [] - for line in open(watch): - try: - rec = json.loads(line) - except ValueError: # the last line can be cut where the watcher was stopped - continue - row = next((c for c in rec.get("cells") or [] if c["id"] == cell), None) - if row and row["head"] != seen: - seen = row["head"] - out.append((row["head"], row["durable_at"])) - return out - - -def join(turns, changes): - index = {t["id"]: t["idx"] for t in turns} - done, out = -1, [] - for head, at in changes: - upto = index.get(head) - if upto is None: - continue - out += [(t, at) for t in turns if done < t["idx"] <= upto] - done = max(done, upto) - return out, [t for t in turns if t["idx"] > done] - - -if __name__ == "__main__": - turns = json.load(open(sys.argv[1])) - joined, lost = join(turns, head_changes(sys.argv[2], sys.argv[3])) - samples = [at - t["sealed_ms"] for t, at in joined] - print(json.dumps({"samples_ms": samples, "turns": len(turns), - "durable": len(joined), "not_durable": [t["id"] for t in lost]})) diff --git a/scripts/measure/hosted/report.py b/scripts/measure/hosted/report.py deleted file mode 100644 index a9ec55ca8b..0000000000 --- a/scripts/measure/hosted/report.py +++ /dev/null @@ -1,103 +0,0 @@ -#!/usr/bin/env python3 -"""The tables of the hosted-relay validation, from the passes' results.json files. - - report.py [card-test.json] - -Prints Markdown: the step table (every pass, min and max, the target beside them and a verdict), the save -latency per tool call, the push samples and the integrity checks. The prose around the tables is written -by hand in docs/BENCH-MOVE.md; this file only does the arithmetic, so no number is typed twice. -""" -import json -import os -import statistics -import sys - - -def load(root): - out = [] - n = 1 - while os.path.exists(os.path.join(root, f"pass{n}", "results.json")): - out.append(json.load(open(os.path.join(root, f"pass{n}", "results.json")))) - n += 1 - return out - - -def fmt_s(x): - return f"{x:.1f} s" if x >= 10 else f"{x:.2f} s" - - -def row(name, vals, target, ok, fmt=fmt_s, note=""): - cells = " | ".join(fmt(v) for v in vals) - verdict = "-" if target == "n/a" else "met" if ok(max(vals)) else "MISSED" - return f"| {name} | {cells} | {fmt(min(vals))} to {fmt(max(vals))} | {target} | {verdict} | {note} |" - - -def steps(passes): - ps = [f"pass {p['pass']}" for p in passes] - head = f"| Step | {' | '.join(ps)} | min to max | target | verdict | what is timed |\n|---|{'---|' * len(ps)}---|---|---|---|" - f = lambda k: [p[k] for p in passes] - seal = lambda who: [statistics.median(c["seal_ms"] for c in p[who]["calls"][1:] if c["seal_ms"] is not None) / 1000 for p in passes] - seal_max = lambda who: [max(c["seal_ms"] for c in p[who]["calls"][1:]) / 1000 for p in passes] - dur = lambda who: [statistics.median(c["durable_ms"] for c in p[who]["calls"] if c["durable_ms"] and c["durable_ms"] < 20000) / 1000 for p in passes] - push = lambda k: [statistics.median(p[k]) / 1000 for p in passes] - rows = [ - row("Pair (scripted)", f("pair_s"), "n/a", lambda v: True, note="the new machine runs `codeaf pair` and shows a link, the first machine runs `codeaf pair approve ` and `y` is typed against the check number"), - row("First upload durable, after the turn ended", f("first_upload_durable_s"), "n/a", lambda v: True, note="202 MB tree, 226 frame puts"), - row("Save: seal after a tool call, A (median)", seal("a_latency"), "~0.15 s", lambda v: v <= 0.16, note="tool call ended to turn sealed"), - row("Save: seal after a tool call, A (worst)", seal_max("a_latency"), "~0.15 s", lambda v: v <= 0.16), - row("Save: seal after a tool call, B (median)", seal("b_latency"), "~0.15 s", lambda v: v <= 0.16), - row("Save: seal to durable at the relay, A (median)", dur("a_latency"), "n/a", lambda v: True, note="calls after the first upload; relay clock within about 0.2 s"), - row("Save: seal to durable at the relay, B (median)", dur("b_latency"), "n/a", lambda v: True), - row("Push: durable at the relay to a new row on B's home (median of 5)", push("push_durable_to_home_ms"), "~0.1 s", lambda v: v <= 0.12, note="new chat on A; screen polled every 10 ms on B"), - row("Push: seal on A to the row on B's home (median of 5)", push("push_seal_to_home_ms"), "~0.1 s", lambda v: v <= 0.12, note="includes the upload"), - row("Cold take, M, B: confirm to chat open", f("cold_take_s"), "<10 s", lambda v: v < 10, note="home screen, `continue here`, 200.7 MB, 209 GETs"), - row("Warm take, M, B, 1 file changed", [p["warm_take_ms"] / 1000 for p in passes], "<5 s", lambda v: v < 5, note="same take as the home screen, `vdemo-take`"), - row("Take back, M, A, 1 file changed on the other side", [p["take_back_ms"] / 1000 for p in passes], "<5 s", lambda v: v < 5, note="same take as the home screen, `vdemo-take`"), - ] - return "\n".join([head] + rows) - - -def latency(passes): - lines = ["| Pass | Machine | Seal after each tool call (ms) | Durable at the relay (ms) |", "|---|---|---|---|"] - for p in passes: - for who, name in (("a_latency", "A"), ("b_latency", "B")): - c = p[who]["calls"] - lines.append(f"| {p['pass']} | {name} | {', '.join(str(x['seal_ms']) for x in c)} | {', '.join(str(x['durable_ms']) for x in c)} |") - return "\n".join(lines) - - -def push(passes): - lines = ["| Pass | Durable to home (ms), five new chats | Seal to home (ms) |", "|---|---|---|"] - for p in passes: - lines.append(f"| {p['pass']} | {', '.join(map(str, p['push_durable_to_home_ms']))} | {', '.join(map(str, p['push_seal_to_home_ms']))} |") - return "\n".join(lines) - - -def integrity(passes): - lines = ["| Check | " + " | ".join(f"pass {p['pass']}" for p in passes) + " |", "|---|" + "---|" * len(passes)] - yes = lambda b: "PASS" if b else "FAIL" - checks = [ - ("`diff -r --no-dereference` A tree against B tree (`.git`, `.furrow`, `node_modules` excluded): no output", lambda i: yes(i["diff_r_exit"] == 0 and i["diff_r_lines"] == 0)), - ("sha256 and mode of every file equal (`treehash.py`)", lambda i: yes(i["treehash_equal_ignoring_node_modules"] and i["modes_equal"])), - ("`.env` 0600, `svc/api/.env` 0640, `web/client/app/.env.local` 0600, `run.sh` 0755 on B, bytes equal", lambda i: yes(all(s["a"] == s["b"] and s["bytes_equal"] for s in i["secrets"].values())) + " (" + "/".join(s["b"] for s in i["secrets"].values()) + ")"), - ("withheld `node_modules` absent on B after the take", lambda i: yes(i["node_modules_on_b_before_rebuild"] == "no")), - ("`git status` on B differs from A's only by the withheld folder", lambda i: yes(i["git_status_only_on_a_before_rebuild"] == ["?? node_modules/"])), - ("`git log` and `git diff` hash equal", lambda i: yes(i["git_log_and_diff_equal"])), - ("no `.codeaf/` in `git status` on either machine", lambda i: yes(not i["codeaf_in_git_status_b"])), - ("`npm ci` on B rebuilds it (seconds in the row below) and `git status` then equals A's", lambda i: yes(i["git_equal_after_rebuild"])), - ("rebuilt `node_modules` has the same files and hashes as A's", lambda i: yes(i["node_modules_rebuilt_equal"] and not i["node_modules_file_diff"])), - ("rebuild time", lambda i: f"{i['rebuild_s']} s"), - ] - for name, fn in checks: - lines.append(f"| {name} | " + " | ".join(fn(p["integrity"]) for p in passes) + " |") - return "\n".join(lines) - - -if __name__ == "__main__": - passes = load(sys.argv[1]) - for title, body in (("Steps", steps(passes)), ("Save latency per tool call", latency(passes)), ("Push samples", push(passes)), ("Integrity", integrity(passes))): - print(f"### {title}\n\n{body}\n") - if len(sys.argv) > 2: - card = json.load(open(sys.argv[2])) - shown = sum(1 for c in card if c["card"]) - print(f"### Resume card frequency\n\n{shown} of {len(card)} cold takes of small chats showed the card; facts present in {sum(1 for c in card if c['withheld'] or c['running'])}.\n") diff --git a/scripts/measure/hosted/setup.sh b/scripts/measure/hosted/setup.sh deleted file mode 100644 index 24d47cd69c..0000000000 --- a/scripts/measure/hosted/setup.sh +++ /dev/null @@ -1,22 +0,0 @@ -#!/usr/bin/env bash -# setup.sh [relay] : clean rig home under on one machine, config with every role pinned, env.sh -# against the relay under test (the second argument, else $CODEAF_RELAY). -set -euo pipefail -R=${1:?usage: setup.sh [relay]} -MODEL=deepseek/deepseek-v4.1-flash -URL=${2:-${CODEAF_RELAY:?CODEAF_RELAY is not set: it is the relay under test, for example https://relay.example.com}} -cfg() { - printf '{\n "approval.guardian": "off",\n "daily_budget_usd": 500,\n "model_pool": "off",\n' - for k in model.plan model.scribe model.talk model.verify model.work models.fallbacks models.tiers.high models.tiers.low models.tiers.mastermind models.tiers.reflex models.tiers.worker; do printf ' "%s": "%s",\n' "$k" "$MODEL"; done - printf ' "setup_seen_at": "2026-09-29T00:00:00Z",\n "task.max_load": 200,\n "telemetry": false,\n "tools.approvalMode": "allow"\n}\n' -} -envtxt() { cat < "$R/home/config.json"; envtxt > "$R/env.sh" -echo ready "$R" diff --git a/scripts/measure/hosted/treehash.py b/scripts/measure/hosted/treehash.py deleted file mode 100644 index e3d682d42f..0000000000 --- a/scripts/measure/hosted/treehash.py +++ /dev/null @@ -1,32 +0,0 @@ -#!/usr/bin/env python3 -"""One task copy, described the same way on either machine: every file's sha256, permission -bits and path (the .git and .furrow entries skipped), then the copy's git state. - - treehash.py -""" -import hashlib -import os -import subprocess -import sys - -root = sys.argv[1] -for base, dirs, files in os.walk(root): - dirs[:] = sorted(d for d in dirs if d not in (".git", ".furrow")) - for f in sorted(files): - if f in (".git", ".furrow"): - continue - p = os.path.join(base, f) - if os.path.islink(p): - continue - print(hashlib.sha256(open(p, "rb").read()).hexdigest(), "%o" % (os.stat(p).st_mode & 0o7777), os.path.relpath(p, root)) - - -def git(*a): - r = subprocess.run(["git", "-C", root, *a], capture_output=True, text=True) - return (r.stdout + r.stderr).rstrip() - - -print("--- status"); print(git("status", "--porcelain")) -print("--- log"); print(git("log", "--oneline", "-8")) -print("--- branch"); print(git("branch", "--show-current")) -print("--- diff sha"); print(hashlib.sha256(git("diff").encode()).hexdigest()) diff --git a/scripts/measure/hosted/tuidrive.py b/scripts/measure/hosted/tuidrive.py deleted file mode 100644 index c626a0d50c..0000000000 --- a/scripts/measure/hosted/tuidrive.py +++ /dev/null @@ -1,265 +0,0 @@ -#!/usr/bin/env python3 -"""Drive the real `codeaf chat` TUI inside one tmux server, on whichever machine this runs on. - -It runs ON the machine whose screen it reads, so a keystroke and the poll that watches its effect share -one clock and one process: no ssh hop sits inside a measured interval. Every timestamp is that -machine's own wall clock in ms; the orchestrator converts B's to A's with a measured offset. -Adapted from scripts/measure/bstep.py, which reads the same home screen, with two changes: a take -works from a chat that is running elsewhere (the card says "running on "), and the same file -serves both machines (VD_TMUX names the tmux server). - - tuidrive.py cap print the pane - tuidrive.py home go to the home screen; JSON rows - tuidrive.py rows JSON list of the home sessions rows - tuidrive.py send [secs] type a prompt, wait for the chat to go busy then idle; JSON timeline - tuidrive.py take [hint] take the first remote row (or the one naming hint); JSON timeline - tuidrive.py open-local open this machine's own first chat from home - tuidrive.py watch-new JSONL: the instant each new row appears on the home screen -""" -import json -import os -import re -import subprocess -import sys -import time - -SESS = os.environ.get("VD_TMUX", "vdb") -ANSI = re.compile(r"\x1b\[[0-9;]*m") - - -def now_ms(): - return int(time.time() * 1000) - - -def tmux(*args): - return subprocess.run(["tmux", "-L", SESS, *args], capture_output=True, text=True).stdout - - -def cap(colour=False): - return tmux("capture-pane", "-p", *(["-e"] if colour else []), "-t", SESS) - - -def key(*keys): - tmux("send-keys", "-t", SESS, *keys) - - -def plain(line): - return ANSI.sub("", line) - - -def wait_for(pred, timeout=20.0, step=0.015): - end = time.time() + timeout - while time.time() < end: - text = cap() - if pred(text): - return now_ms(), text - time.sleep(step) - return None, cap() - - -def on_home(text): - return any(l.strip().startswith("sessions") for l in text.splitlines()) - - -def home_rows(text): - """The text of each row under the `sessions` header, up to the next section.""" - out, seen = [], False - for raw in text.splitlines(): - p = plain(raw).rstrip() - if p.strip().startswith("sessions"): - seen = True - continue - if not seen: - continue - if not p.strip() or p.lstrip().startswith("─") or "type to search" in p or p.strip().startswith(("spend", "projects")): - break - out.append(p) - return out - - -def is_remote(row): - """A chat on this machine carries a dim bullet before its name; a chat on another machine has none. - The status sentence cannot tell them apart, because a long name is cut before the sentence is drawn.""" - return not row.lstrip().startswith("·") - - -def go_home(): - """Space-space is the footer's way home from a chat; at home it is not sent.""" - for _ in range(4): - if on_home(cap()): - return True - key("Space", "Space") - time.sleep(0.8) - return on_home(cap()) - - -def selected_row(): - """The cursor row's text, found by the background the sessions panel gives it.""" - for raw in cap(True).splitlines(): - p = plain(raw).strip() - if "48;5;235m" in raw and p and not p.startswith("sessions"): - return p - return "" - - -def select(pred, limit=20): - for direction in ("Down", "Up"): - for _ in range(limit): - sel = selected_row() - if sel and pred(sel): - return sel - key(direction) - time.sleep(0.12) - return "" - - -# A chat that left a dev server running says `1 job working` in its footer for as long as the server lives, though the -# turn is over: the word `idle` never comes back, so a running job alone must not keep a finished turn from ending. -JOBS_ONLY = re.compile(r"\d+ jobs? +working$") - - -def status_idle(text): - """The footer's right-hand word, `idle` once the chat waits for the next prompt (or only background jobs run).""" - return any(l.startswith("─") and (l.rstrip(" ─").endswith("idle") or JOBS_ONLY.search(l.rstrip(" ─"))) - for l in map(plain, text.splitlines())) - - -def finished_after(text, token): - """Whether the turn that began with a prompt naming token has ended: its `worked` line is on screen - below the prompt and the footer says idle. The screen scrolls, so a count of lines would not do.""" - lines = [plain(l) for l in text.splitlines()] - foot = next((i for i in range(len(lines) - 1, -1, -1) if lines[i].startswith("─ ")), len(lines)) - seen = [i for i, l in enumerate(lines[:foot]) if token in l] - return bool(seen) and status_idle(text) and any("▸ worked" in l for l in lines[seen[-1]:foot]) - - -def fresh_word(prompt, screen): - """A word of the prompt that is not on screen yet, so the screen can tell this turn from the last.""" - words = [w for w in re.findall(r"[A-Za-z0-9][A-Za-z0-9._-]{4,}", prompt) if w not in screen] - dashed = [w for w in words if "-" in w] # the probes name themselves with a dashed word; prose words also sit in the footer tips - return (dashed or words or [prompt[:30]])[-1] - - -def send(prompt, secs): - token = fresh_word(prompt, cap()) - t = {"typed_ms": now_ms(), "token": token} - key("-l", prompt) - time.sleep(0.3) - key("Enter") - t["enter_ms"] = now_ms() - t["idle_ms"], text = wait_for(lambda x: finished_after(x, token), secs) - t["pane"] = text - return t - - -def title_of(row): - return re.sub(r"\s{2,}.*", "", row.strip()).strip("·▸ ") - - -def local_titles(text): - return {title_of(r) for r in home_rows(text) if not is_remote(r)} - - -def chat_open(text): - """A chat is open when the tab strip under the header names it with a close mark.""" - return "×" in "\n".join(text.splitlines()[:4]) and not on_home(text) - - -def take(hint): - """Continue the first remote chat (or the one naming hint) here. - - The home screen opens the chat by itself the moment the take ends, and raises the resume card in - it when the chat left something behind. So the take ends when the chat is on screen; the card is - read there, and answered with esc (`not now`) so a later prompt is typed into the chat and not the card. When - the chat does not open by itself, it is found again as the one local row that was not there before - (a chat is renamed once its first turn is named) and opened with enter.""" - t = {} - go_home() - before = local_titles(cap()) - row = select(lambda s: is_remote(s) and (hint == "-" or hint.lower() in s.lower())) - if not row: - return {"error": "no remote row", "pane": cap()} - t["row"] = row - key("Enter") - t["card_ms"], card = wait_for(lambda x: "Continue this chat here?" in x, 10) - t["card"] = [plain(l).strip(" │") for l in card.splitlines() if "durable turn" in l] - key("1") - time.sleep(0.15) - t["confirm_ms"] = now_ms() - key("Enter") - t["taken_ms"], pane = wait_for(lambda x: "Continue this chat here?" not in x and (chat_open(x) or bool(local_titles(x) - before)), 3600, 0.02) - t["listed_ms"] = t["taken_ms"] # the take is over: the chat is open, or listed on home - t["auto_opened"] = chat_open(pane) - if not chat_open(pane): - new = sorted(local_titles(pane) - before) - if not new: - return dict(t, error="no new local row after take", pane_after_take=cap()) - t["row_after_take"] = new[0] - select(lambda s: title_of(s) == new[0], limit=10) - t["open_key_ms"] = now_ms() - key("Enter") - t["taken_ms"], pane = wait_for(chat_open, 120) - t["open_ms"] = t["taken_ms"] - t["first_screen"] = pane - t["resume_card_ms"], shown = wait_for(lambda x: "not now" in x and "set up" in x, 4) - t["resume_card"] = shown if t["resume_card_ms"] else "" - t["ready_ms"], _ = wait_for(lambda x: "›" in x and "type to search" not in x, 60) - if t["resume_card_ms"]: - # The card's own `not now`. An escape sent in the instant the card is drawn is not read by it (a person cannot - # type that fast), so the key is sent again until the card is gone. - for _ in range(4): - time.sleep(0.7) - key("Escape") - if wait_for(lambda x: "esc not now" not in x, 2)[0]: - break - t["screen_after"] = cap() - return t - - -def open_local(): - """Open this machine's own first chat from the home screen (the `here` rows, not `new conversation`).""" - go_home() - row = select(lambda s: not is_remote(s) and "new conversation" not in s) - if not row: - return {"error": "no local chat row", "pane": cap()} - t = {"row": row, "open_key_ms": now_ms()} - key("Enter") - t["open_ms"], first = wait_for(lambda x: "×" in "\n".join(x.splitlines()[:4]), 60) - t["ready_ms"], _ = wait_for(lambda x: "›" in x and "type to search" not in x, 60) - t["screen"] = cap() - return t - - -def watch_new(seconds, step_ms): - """Print the instant a title gains a row on the home screen. Titles repeat (two chats started from - the same kind of prompt are named alike), so rows are counted, not looked up.""" - from collections import Counter - end = time.time() + seconds - last = None - while time.time() < end: - rows = home_rows(cap()) - now = Counter(title_of(r) for r in rows) - if last is not None: - for ttl, n in now.items(): - if ttl != "new conversation" and n > last.get(ttl, 0): - print(json.dumps({"ms": now_ms(), "new": ttl, "rows": len(rows)}), flush=True) - last = now - time.sleep(step_ms / 1000) - - -if __name__ == "__main__": - cmd, args = sys.argv[1], sys.argv[2:] - if cmd == "cap": - print(cap()) - elif cmd == "home": - print(json.dumps({"home": go_home(), "rows": home_rows(cap())})) - elif cmd == "rows": - print(json.dumps(home_rows(cap()))) - elif cmd == "send": - print(json.dumps(send(args[0], float(args[1]) if len(args) > 1 else 180))) - elif cmd == "take": - print(json.dumps(take(args[0] if args else "-"))) - elif cmd == "open-local": - print(json.dumps(open_local())) - elif cmd == "watch-new": - watch_new(float(args[0]), float(args[1])) diff --git a/scripts/measure/hosted/vdemotake.go.txt b/scripts/measure/hosted/vdemotake.go.txt deleted file mode 100644 index d9135c398a..0000000000 --- a/scripts/measure/hosted/vdemotake.go.txt +++ /dev/null @@ -1,43 +0,0 @@ -package main - -// The validation rig's take verb: `codeaf vdemo-take ` runs the same take the home screen's -// `continue here` runs (the same Continuer, built the way synctake.go builds it) and prints one JSON -// line. The home screen draws no remote row for a chat this machine already lists, so a warm take and -// a take back cannot be pressed there; this is the only way to time them through the product's own -// code. NOT product code: the rig copies this file into a private build of the tree, never commits it. - -import ( - "context" - "encoding/json" - "fmt" - "os" - "time" - - "github.com/Agent-Field/codeaf/internal/cellstore" - "github.com/Agent-Field/codeaf/internal/syncsetup" -) - -func init() { - if len(os.Args) != 3 || os.Args[1] != "vdemo-take" { - return - } - out := map[string]any{} - s, err := syncOf() - if err == nil { - cont := s.Continuer(cellstore.EngineFor(""), syncsetup.TakeOptions{DeviceName: deviceName(), RootFor: takeRootFor, Notify: func(string) {}}) - t0 := time.Now() - got, terr := cont.Take(context.Background(), os.Args[2]) - out["take_ms"] = float64(time.Since(t0).Microseconds()) / 1000 - err = terr - if terr == nil { - out["root"], out["kept"], out["resume"] = got.Taken.Cell.Root, got.Taken.Kept, got.Resume - out["fence"] = got.Taken.Driving.Fence - } - } - if err != nil { - out["error"] = err.Error() - } - b, _ := json.Marshal(out) - fmt.Println(string(b)) - os.Exit(0) -} diff --git a/scripts/relay-bill.sh b/scripts/relay-bill.sh deleted file mode 100755 index aea78e9dc6..0000000000 --- a/scripts/relay-bill.sh +++ /dev/null @@ -1,41 +0,0 @@ -#!/usr/bin/env bash -# One heavy user against the relay under test (CODEAF_RELAY), then what Cloudflare billed for it. -# -# BILL_TITLE="before (10c125f4e)" scripts/relay-bill.sh before [idle-minutes] -# BILL_TITLE="after push (845e5061d)" scripts/relay-bill.sh after [idle-minutes] -# The label names the run's files; every labelled run so far is a column of the table, oldest first. -# -# It drives the real client (internal/relaybill through the home screen's own -# pace in internal/tui3), waits for the analytics to settle, reads the billed -# usage of each phase, and rewrites section 9 of the decision document with the -# one column per run. A run takes about idle-minutes -# plus ten, so start it detached: -# setsid nohup scripts/relay-bill.sh after > after.log 2>&1 & -# Use the same idle length for every run: the extrapolation is per second. -set -euo pipefail - -label=${1:?usage: relay-bill.sh label [idle-minutes]} -idle=${2:-20} -url=${CODEAF_RELAY:?CODEAF_RELAY is not set: it is the relay under test, for example https://relay.example.com} -runs=${BILL_DIR:-$HOME/relay-bill-runs} -doc=${BILL_DOC:?BILL_DOC is not set: it is the decision document whose billing section this run rewrites} -root=$(cd "$(dirname "$0")/.." && pwd) -cost=$root/relay/hosted/cost -mkdir -p "$runs" -# The column's heading in section 9, for example BILL_TITLE="after push (845e5061d)". -echo "${BILL_TITLE:-$label}" >"$runs/$label.title" - -cd "$root" -make furrow >/dev/null -go test -tags relaybill ./internal/tui3 -run TestRelayBill -count=1 -timeout 120m -v \ - -args -bill-url "$url" -bill-out "$runs/$label.run.json" -bill-idle "${idle}m" -python3 "$cost/billed.py" --manifest "$runs/$label.run.json" --out "$runs/$label.billed.json" - -# Every run so far is a column, oldest first. -args=() -for f in $(ls -tr "$runs"/*.run.json); do - name=$(basename "$f" .run.json) - [[ -s $runs/$name.billed.json ]] && args+=(--column "$(cat "$runs/$name.title")" "$f" "$runs/$name.billed.json") -done -python3 "$cost/bill.py" "${args[@]}" --doc "$doc" -echo "EXIT=0" diff --git a/scripts/relay-conformance.sh b/scripts/relay-conformance.sh index 399f6ea427..64de4d66d7 100755 --- a/scripts/relay-conformance.sh +++ b/scripts/relay-conformance.sh @@ -6,7 +6,7 @@ # # Two relays, the Go one and the Worker, are run through the same suites: in process for the Go relay, # as a live process for both. The Worker runs under `wrangler dev --local` on free ports and never -# against staging, so a billing window stays clean. Tracks run sequentially on the shared box; +# against a deployed relay. Tracks run sequentially on the shared box; # CONFORMANCE_PARALLEL=1 opts into separate concurrent tracks. Their suites run one after another, because the watch suite's thousand # sockets would otherwise stall a Worker that a lease case is waiting on. Cases default to two # at a time (watch cases run alone), and the local Worker transport bounds requests, retrying only the