Follow-ups to hosted storage opt-in: tools stop when a chat moves, closing sentences arrive, moved chats keep their words

Makefile

@@ -218,6 +218,7 @@ test-tooling:
 	bash scripts/shard-test_test.sh
 	bash scripts/relay-fingerprint_test.sh
 	bash scripts/relay-wrangler-config_test.sh
+	bash scripts/raft-proof-footprint_test.sh
 	bash scripts/relay-deploy_test.sh
 	bash bench/canary/lib/repo_test.sh
 

cmd/codeaf/chatv3_layout.go

@@ -789,6 +789,12 @@ func v3Seated(cfg session.Config) session.Config {
 	seat, machine, seals := v3SeatWith(cfg.Place, nil, syncDrives)
 	cfg.Seat, cfg.Interrupted = seat, executor.InterruptedOn(seat)
 	cfg.Renewals = syncDrives.renewalsOf(cfg.Place.ID())
+	cfg.FinishSaving, cfg.SavingNotices = nil, nil
+	if drive := syncDrives.renewalsOf(cfg.Place.ID()); drive != nil {
+		book, id := syncDrives, cfg.Place.ID()
+		cfg.FinishSaving = func() error { return book.closeCell(id) }
+		cfg.SavingNotices = drive.TakeSavingNotices
+	}
 	if machine != nil {
 		cfg.Machine = machine
 	}

cmd/codeaf/chatv3_pair.go

@@ -39,6 +39,9 @@ func (chatPairDoor) Offer(ctx context.Context, ui pair.OfferUI) (string, error)
 		return "", explainSyncUnreachable(err, route.URL)
 	}
 	label, err := pair.Offer(ctx, route, grant, chatOfferUI{OfferUI: ui, service: route.Shown})
+	if err == nil {
+		samplePairedStorage(ctx, home.Dir(), route)
+	}
 	return label, explainSyncUnreachable(err, route.URL)
 }
 
@@ -82,7 +85,9 @@ func (chatPairDoor) JoinReplacing(ctx context.Context, typed string, ui pair.Joi
 	}
 	joined, err := pair.Join(ctx, route, pairJoining(true), typed, ui)
 	if err == nil {
+		waitStorage := startPairedStorage(ctx, home.Dir(), route)
 		err = ownDeviceRecord(ctx, home.Dir(), route)
+		waitStorage()
 	}
 	return joined, explainSyncUnreachable(err, route.URL)
 }

cmd/codeaf/chatv3_surface.go

@@ -87,6 +87,13 @@ func runSurface(ctx context.Context, options tui3.Options) error {
 			previousSaveFailure(err)
 		}
 	}
+	previousNotices := options.SavingNotices
+	options.SavingNotices = func(lines []string) {
+		recordFinalSaveNotices(lines)
+		if previousNotices != nil {
+			previousNotices(lines)
+		}
+	}
 	previousClosing, previousForce := options.ClosingStarted, options.ForceQuit
 	options.ClosingStarted = func(done <-chan struct{}) {
 		retainFinalClose(done)

cmd/codeaf/final_publish_notice_test.go

@@ -0,0 +1,171 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Agent-Field/codeaf/internal/blobstore"
+	"github.com/Agent-Field/codeaf/internal/cellstore"
+	"github.com/Agent-Field/codeaf/internal/chatlist"
+	"github.com/Agent-Field/codeaf/internal/directory"
+	"github.com/Agent-Field/codeaf/internal/home"
+	"github.com/Agent-Field/codeaf/internal/identity"
+	"github.com/Agent-Field/codeaf/internal/remote"
+	"github.com/Agent-Field/codeaf/internal/reqsign"
+	"github.com/Agent-Field/codeaf/internal/session"
+	"github.com/Agent-Field/codeaf/internal/syncsetup"
+	"github.com/Agent-Field/codeaf/internal/tui3"
+)
+
+func TestFinalPublishDisplacementReachesTheRestoredWindowOnce(t *testing.T) {
+	for _, road := range []string{"engine", "no-host"} {
+		for _, displaced := range []bool{true, false} {
+			t.Run(fmt.Sprintf("%s/displaced=%t", road, displaced), func(t *testing.T) {
+				t.Setenv("CODEAF_NO_UPDATE_CHECK", "1")
+				cfg, c := spokenDoorChat(t)
+				t.Cleanup(cellstore.StopSpawned)
+				t.Cleanup(stopFinalSaveProgress)
+				id, err := identity.Ensure(home.Dir())
+				if err != nil {
+					t.Fatal(err)
+				}
+				dir := directory.NewMemory(time.Now)
+				defer dir.Close()
+				key := directory.MetadataKey(id.CellKey())
+				name, err := directory.SealName(key, "desk")
+				if err != nil {
+					t.Fatal(err)
+				}
+				if err := dir.For("other-device").PutDevice(t.Context(), "other-device", directory.Device{V: 1, Name: name}); err != nil {
+					t.Fatal(err)
+				}
+				auth := reqsign.AuthenticateAt(time.Now)
+				mux := http.NewServeMux()
+				mux.Handle("/v1/dir/", directory.Handler(auth, func(string) (directory.Directory, error) { return dir, nil }))
+				frames := blobstore.NewMemory()
+				mux.Handle("/", blobstore.Handler(auth, func(string) (blobstore.Store, error) { return frames, nil }))
+				var closing, discovered atomic.Bool
+				server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					if displaced && closing.Load() && strings.HasSuffix(r.URL.Path, "/publish") && discovered.CompareAndSwap(false, true) {
+						if _, err := dir.For("other-device").Acquire(r.Context(), c.ID, directory.AcquireOpts{Force: true}); err != nil {
+							t.Error(err)
+						}
+					}
+					mux.ServeHTTP(w, r)
+				}))
+				defer server.Close()
+				t.Setenv(syncsetup.URLVar, server.URL)
+				t.Setenv(syncsetup.IntervalVar, "3600000")
+				if err := syncsetup.SaveStorage(home.Dir(), server.URL, id.ID(), "on"); err != nil {
+					t.Fatal(err)
+				}
+				cfg = v3Seated(cfg)
+				watch := cfg.Seals.(*cellstore.SealWatch)
+				watch.OnTail, watch.OnTurnEnd = nil, nil
+				d := syncDrives.renewalsOf(c.ID)
+				startupFinished(d)
+				if d.started() == nil {
+					t.Fatal("writer did not start")
+				}
+				if _, err := d.started().Store(cellstore.EngineFor(cfg.Workspace)).Seal(t.Context(), c, cellstore.TurnInfo{}); err != nil {
+					t.Fatal(err)
+				}
+				d.Idle()
+				waitStorageOffBoundary(t, "initial publish", func() bool {
+					view, err := dir.For("observer").Cell(t.Context(), c.ID)
+					return err == nil && view.Cell.Head != ""
+				})
+				agent, err := session.New(cfg)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer agent.Close()
+				events, err := agent.SubmitBash(t.Context(), "! printf final-work > final-work")
+				if err != nil {
+					t.Fatal(err)
+				}
+				for event := range events {
+					if event.Kind == session.EventError {
+						t.Fatal(event.Text)
+					}
+				}
+				var window tui3.Agent = agent
+				if road == "engine" {
+					loop, err := remote.Loopback(remote.Hello{}, remote.Options{Boot: func(remote.Hello) (*remote.Engine, error) {
+						return &remote.Engine{Agent: agent, SessionFile: cfg.SessionFile, Workspace: cfg.Workspace, FinishSaving: func(remote.WrappedAgent) error { return syncDrives.closeCell(c.ID) }}, nil
+					}})
+					if err != nil {
+						t.Fatal(err)
+					}
+					defer loop.Close()
+					window = loop.Client.Agent()
+				}
+				input, typed := io.Pipe()
+				defer input.Close()
+				defer typed.Close()
+				screen := &surfaceFrameWriter{ready: make(chan struct{})}
+				previousRun := runSurfaceProgram
+				runSurfaceProgram = func(ctx context.Context, options tui3.Options) error {
+					options.Input, options.Output = input, screen
+					options.UpdateCheck = nil
+					return tui3.Run(ctx, options)
+				}
+				defer func() { runSurfaceProgram = previousRun }()
+				ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second)
+				defer cancel()
+				done := make(chan error, 1)
+				go func() {
+					done <- runSurface(ctx, tui3.Options{Agent: window, SessionFile: cfg.SessionFile, Workspace: cfg.Workspace, ProfileDir: home.Dir(), Seal: sealSeamOf(window, cfg.Seals), Input: input, Output: screen, Width: 160, Height: 35, Linear: true})
+				}()
+				select {
+				case <-screen.ready:
+				case err := <-done:
+					t.Fatalf("window closed before first frame: %v", err)
+				case <-ctx.Done():
+					t.Fatal("window never painted")
+				}
+				closing.Store(true)
+				if _, err := io.WriteString(typed, "/quit\r"); err != nil {
+					t.Fatal(err)
+				}
+				if err := <-done; err != nil {
+					t.Fatal(err)
+				}
+				if displaced && !discovered.Load() {
+					t.Fatal("final publish did not encounter displacement")
+				}
+				oldOut, oldFlush := os.Stdout, flushFinalSaves
+				out, err := os.CreateTemp(t.TempDir(), "receipt")
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer out.Close()
+				os.Stdout = out
+				flushFinalSaves = func(context.Context) error { return nil }
+				defer func() { os.Stdout, flushFinalSaves = oldOut, oldFlush }()
+				_, _ = io.WriteString(out, "screen restored\n")
+				_ = finishFinalSaving(t.Context())
+				_ = finishFinalSaving(t.Context())
+				raw, err := os.ReadFile(out.Name())
+				if err != nil {
+					t.Fatal(err)
+				}
+				count := strings.Count(string(raw), chatlist.Superseded("desk"))
+				if displaced && count != 1 || !displaced && count != 0 {
+					t.Fatalf("restored receipt: %q", raw)
+				}
+				if !displaced && string(raw) != "screen restored\n" {
+					t.Fatalf("clean final save added a sentence: %q", raw)
+				}
+			})
+		}
+	}
+}

cmd/codeaf/final_save.go

@@ -31,6 +31,7 @@ const quitSavedHereWord = syncwords.SavedHerePending
 var finalSaving struct {
 	sync.Mutex
 	failure       error
+	notices       []string
 	surface       bool
 	finishing     bool
 	address       string
@@ -52,6 +53,7 @@ func armFinalSaving() chan struct{} {
 	finalSaving.Lock()
 	defer finalSaving.Unlock()
 	finalSaving.surface = true
+	finalSaving.notices = nil
 	finalSaving.finishing = false
 	finalSaving.address = ""
 	finalSaving.durable = nil
@@ -236,6 +238,17 @@ func finishFinalSaving(ctx context.Context) (ok bool) {
 	}
 	err, surface, finishing, address, checks, legacy := takeFinalReport()
 	defer func() { completeFinalReport(done, ok) }()
+	lines := takeFinalSaveNotices()
+	if len(lines) > 0 {
+		stopFinalSaveProgress()
+		terminal := os.Stderr
+		if surface {
+			terminal = os.Stdout
+		}
+		for _, line := range lines {
+			fmt.Fprintln(terminal, line)
+		}
+	}
 	if err == nil && !finishing {
 		return true
 	}
@@ -469,3 +482,29 @@ func setFinalSyncService(address string) {
 	defer finalSaving.Unlock()
 	finalSaving.address = address
 }
+
+// recordFinalSaveNotices keeps each closing sentence until terminal restoration.
+func recordFinalSaveNotices(lines []string) {
+	finalSaving.Lock()
+	defer finalSaving.Unlock()
+	for _, line := range lines {
+		found := false
+		for _, previous := range finalSaving.notices {
+			if previous == line {
+				found = true
+				break
+			}
+		}
+		if line != "" && !found {
+			finalSaving.notices = append(finalSaving.notices, line)
+		}
+	}
+}
+
+func takeFinalSaveNotices() []string {
+	finalSaving.Lock()
+	defer finalSaving.Unlock()
+	lines := finalSaving.notices
+	finalSaving.notices = nil
+	return lines
+}

cmd/codeaf/pair.go

@@ -213,7 +213,9 @@ func (d pairDoor) show(ctx context.Context, relay string) error {
 	if err != nil {
 		return d.endedByPerson(ctx, err)
 	}
+	waitStorage := startPairedStorage(ctx, d.joining(false).Home, route)
 	d.recordOwnDeviceOrSay(ctx, route)
+	waitStorage()
 	d.term.say(pair.PairedChatsLine(label))
 	return nil
 }

cmd/codeaf/pair_storage.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/Agent-Field/codeaf/internal/blobstore"
+	"github.com/Agent-Field/codeaf/internal/guard"
+	"github.com/Agent-Field/codeaf/internal/identity"
+	"github.com/Agent-Field/codeaf/internal/pair"
+	"github.com/Agent-Field/codeaf/internal/reqsign"
+	"github.com/Agent-Field/codeaf/internal/syncsetup"
+)
+
+const pairedStorageWithin = 1500 * time.Millisecond
+
+// startPairedStorage overlaps the optional reading with the remaining pairing work.
+// Joining before success lets another local reader see a timely answer immediately.
+func startPairedStorage(ctx context.Context, dir string, route pair.Mailbox) func() {
+	ctx, cancel := context.WithTimeout(ctx, pairedStorageWithin)
+	done := make(chan struct{})
+	// A fault in this optional reading must never take pairing or the surface
+	// down, and done still closes so the pairing that waits on it finishes.
+	guard.Go("pair/storage", func() {
+		defer close(done)
+		defer cancel()
+		samplePairedStorage(ctx, dir, route)
+	})
+	return func() { <-done }
+}
+
+// samplePairedStorage makes a completed pairing visible to other local windows.
+// An absent field or failed request must preserve the previous saved answer.
+func samplePairedStorage(ctx context.Context, dir string, route pair.Mailbox) {
+	id, err := identity.Load(dir)
+	if err != nil {
+		return
+	}
+	dev, err := identity.Device(dir)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(ctx, pairedStorageWithin)
+	defer cancel()
+	store := blobstore.NewHTTP(route.URL, reqsign.SignFor(homeSigner{id, dev}, time.Now), &http.Client{})
+	counts, err := store.Stats(ctx)
+	if err == nil && ctx.Err() == nil && counts.Storage != "" {
+		_ = syncsetup.SaveStorage(dir, route.URL, id.ID(), counts.Storage)
+	}
+}