From c1ec418c3fd30922ca1ae54e71883772de840297 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 16:47:20 -0400 Subject: [PATCH 1/4] chat: a chat another process holds gets no sync drive from here openV3Agent seated a chat, and so started its sync drive, before the session open learned that another process holds the chat's journal. On the same computer the relay lease is this device's, so the drive resumed the other process's fence (a heartbeat and a rewritten receipt), and this process's quit then released that fence and deleted the receipt. The desktop app's chat flipped to "available to continue here" and refused its next send. A --no-host terminal left at "Move this conversation here?" did it on quit; a session host that booted the held chat did it when it stopped or retired. The drive side openV3Agent seats now waits, parked, until open() holds the journal, and is abandoned without any relay call when it does not. A parked drive looks to gates, seals and Close like a startup in progress, so restored work still waits for the ownership decision. Openers of one chat in one process share the parked drive. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/chatv3.go | 4 +- cmd/codeaf/chatv3_layout.go | 30 +- cmd/codeaf/heldchat_drive_test.go | 470 ++++++++++++++++++ cmd/codeaf/syncdrive.go | 56 ++- cmd/codeaf/syncdrive_admission.go | 2 +- cmd/codeaf/syncdrive_live.go | 57 ++- ...0-held-chat-drive-waits-for-its-journal.md | 13 + 7 files changed, 611 insertions(+), 21 deletions(-) create mode 100644 cmd/codeaf/heldchat_drive_test.go create mode 100644 docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md diff --git a/cmd/codeaf/chatv3.go b/cmd/codeaf/chatv3.go index 465dc02a50..13c113852e 100644 --- a/cmd/codeaf/chatv3.go +++ b/cmd/codeaf/chatv3.go @@ -1473,7 +1473,8 @@ func openV3Agent(cfg session.Config, workspace string, open func(session.Config) if model != "" { cfg.Model = model } - cfg = v3Seated(v3Migrated(cfg)) + cfg, claim := v3SeatedPending(v3Migrated(cfg)) + defer claim(false) // Restore the gate before construction, so restored work cannot start behind // the profile default. Return the launch config unchanged: a subsequent new // conversation must not inherit this one's saved override. @@ -1505,6 +1506,7 @@ func openV3Agent(cfg session.Config, workspace string, open func(session.Config) } agent, err := open(restored) if err == nil { + claim(true) if model != "" { if err := agent.ContinueWithModel(model, modelNotice); err != nil { closeV3Agent(agent) diff --git a/cmd/codeaf/chatv3_layout.go b/cmd/codeaf/chatv3_layout.go index 468220e305..7d05494349 100644 --- a/cmd/codeaf/chatv3_layout.go +++ b/cmd/codeaf/chatv3_layout.go @@ -787,16 +787,23 @@ func v3Migrated(cfg session.Config) session.Config { return cfg } -// v3Seated gives the session the executor it owns, built once here from the -// class the cell declared and the workspace its tools run in. +// v3Seated is for a caller that already holds the journal. It gives the session +// the executor it owns, built from the cell class and its tools' workspace. // // A folder that is not a readable cell gets no seat and its calls run on the // host as they always did, and so does every folder without a compatible engine. The // seat is set on every call, so a config reused for the next conversation never // carries the last one's executor. func v3Seated(cfg session.Config) session.Config { + cfg, claim := v3SeatedPending(cfg) + claim(true) + return cfg +} + +// v3SeatedPending builds the seat before committing its drive to journal ownership. +func v3SeatedPending(cfg session.Config) (session.Config, func(bool)) { cfg.Seat, cfg.Machine, cfg.Seals, cfg.Interrupted = nil, nil, nil, nil - seat, machine, seals := v3SeatWith(cfg.Place, nil, syncDrives) + seat, machine, seals, claim := v3SeatPending(cfg.Place, nil, syncDrives) cfg.Seat, cfg.Interrupted = seat, executor.InterruptedOn(seat) cfg.Renewals = syncDrives.renewalsOf(cfg.Place.ID()) if machine != nil { @@ -805,7 +812,7 @@ func v3Seated(cfg session.Config) session.Config { if seals != nil { cfg.Seals = seals } - return cfg + return cfg, claim } // v3SeatOf is the seat of the cell a place names, the machine view it was @@ -822,12 +829,19 @@ func v3SeatOf(place session.Place, say func(string)) (executor.Seat, *preflight. // the chat its tool calls are refused. A nil book, or sync off, is the seat // exactly as it was. func v3SeatWith(place session.Place, say func(string), drives *driveBook) (executor.Seat, *preflight.Machine, *cellstore.SealWatch) { + seat, machine, watch, claim := v3SeatPending(place, say, drives) + claim(true) + return seat, machine, watch +} + +func v3SeatPending(place session.Place, say func(string), drives *driveBook) (executor.Seat, *preflight.Machine, *cellstore.SealWatch, func(bool)) { + noop := func(bool) {} if !cell.Available() || place.Dir == "" || place.Workspace == "" { - return nil, nil, nil + return nil, nil, nil, noop } c, err := cell.OpenAt(place.Dir, filepath.Base(place.Dir)) if err != nil { - return nil, nil, nil + return nil, nil, nil, noop } class, ok := executor.ParseClass(string(c.Meta().Class)) if !ok { @@ -836,7 +850,7 @@ func v3SeatWith(place session.Place, say func(string), drives *driveBook) (execu machine, err := preflight.OpenMachine(c.Root, place.Workspace) watch := &cellstore.SealWatch{Say: say} failed(watch, err) - drive := drives.driveOf(c, cellstore.EngineFor(place.Workspace), watch.Report) + drive, claim := drives.pendingDriveOf(c, cellstore.EngineFor(place.Workspace), watch.Report) if drive != nil { watch.OnTurnEnd = drive.Idle // the agent waits for the person: upload what is sealed now } @@ -847,7 +861,7 @@ func v3SeatWith(place session.Place, say func(string), drives *driveBook) (execu // the first input; their journal writes need a tail seal even before chat. // Window attachment reuses the engine's seat rather than constructing one. watch.Waiting() - return executor.Gated(seat, driveGate(drive), driveWorkGate(drive)), machine, watch + return executor.Gated(seat, driveGate(drive), driveWorkGate(drive)), machine, watch, claim } // sealTail seals what the chat wrote after the turn-end seal (titles, summaries, diff --git a/cmd/codeaf/heldchat_drive_test.go b/cmd/codeaf/heldchat_drive_test.go new file mode 100644 index 0000000000..80824ebd6d --- /dev/null +++ b/cmd/codeaf/heldchat_drive_test.go @@ -0,0 +1,470 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/cell" + "github.com/Agent-Field/codeaf/internal/cell/celltest" + "github.com/Agent-Field/codeaf/internal/cellstore" + "github.com/Agent-Field/codeaf/internal/directory" + "github.com/Agent-Field/codeaf/internal/executor" + "github.com/Agent-Field/codeaf/internal/filelock" + "github.com/Agent-Field/codeaf/internal/home" + "github.com/Agent-Field/codeaf/internal/identity" + "github.com/Agent-Field/codeaf/internal/relayserve" + "github.com/Agent-Field/codeaf/internal/session" + "github.com/Agent-Field/codeaf/internal/syncsetup" +) + +// heldChatRelay counts every cell write, including creation, rather than only +// the heartbeat that first exposed the second opener's ownership side effect. +func heldChatRelay(t *testing.T, id string) (*syncsetup.Sync, *atomic.Int64) { + t.Helper() + service := relayserve.New(relayserve.Config{Store: t.TempDir()}) + writes := &atomic.Int64{} + path := "/v1/dir/cells/" + id + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && (r.URL.Path == path || strings.HasPrefix(r.URL.Path, path+"/")) { + writes.Add(1) + } + service.Handler.ServeHTTP(w, r) + })) + t.Cleanup(func() { server.Close(); _ = service.Close() }) + t.Setenv(syncsetup.URLVar, server.URL) + t.Setenv(syncsetup.IntervalVar, "3600000") + if _, err := identity.Ensure(home.Dir()); err != nil { + t.Fatalf("make this computer's identity: %v", err) + } + if syncsetup.Quiet(home.Dir()) { + t.Fatalf("the existing non-solo identity must permit relay requests") + } + s, ok, err := syncsetup.Open(home.Dir()) + if err != nil || !ok { + t.Fatalf("open relay sync = %v, %v; want true, nil", ok, err) + } + return s, writes +} + +type heldChatReceipt struct { + path string + raw []byte + info os.FileInfo +} + +func snapshotHeldChatReceipt(t *testing.T, s *syncsetup.Sync, id string) heldChatReceipt { + t.Helper() + key := sha256.Sum256([]byte(s.Ledger + "/" + s.Device.ID() + "/" + id)) + path := filepath.Join(home.Dir(), "v3", "sync", "holds", fmt.Sprintf("%x", key)) + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read P1's hold receipt: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat P1's hold receipt: %v", err) + } + return heldChatReceipt{path: path, raw: raw, info: info} +} + +func (before heldChatReceipt) check(t *testing.T) { + t.Helper() + raw, err := os.ReadFile(before.path) + if err != nil { + t.Fatalf("P1's hold receipt after P2 quit = %v; want the original receipt", err) + } + if !bytes.Equal(raw, before.raw) { + t.Fatalf("P1's hold receipt = %q; want unchanged %q", raw, before.raw) + } + info, err := os.Stat(before.path) + if err != nil { + t.Fatalf("stat P1's hold receipt after P2 quit: %v", err) + } else if !os.SameFile(before.info, info) { + t.Fatalf("P2 rewrote P1's hold receipt; want the same file") + } +} + +func checkHeldChatLease(t *testing.T, s *syncsetup.Sync, id string, before directory.CellView) { + t.Helper() + after, err := s.Dir.Cell(t.Context(), id) + if err != nil { + t.Fatalf("read the held chat after P2 quit: %v", err) + } + if after.Cell.Lease.Device != before.Cell.Lease.Device || after.Cell.Lease.Fence != before.Cell.Lease.Fence { + t.Fatalf("lease after P2 quit = %s fence %d; want %s fence %d", after.Cell.Lease.Device, after.Cell.Lease.Fence, before.Cell.Lease.Device, before.Cell.Lease.Fence) + } + if after.Cell.Lease.Expires == 0 || after.Cell.Lease.Expires != before.Cell.Lease.Expires { + t.Fatalf("lease expiry after P2 quit = %d; want unchanged, unreleased %d", after.Cell.Lease.Expires, before.Cell.Lease.Expires) + } + if after.Cell.Head != before.Cell.Head { + t.Fatalf("relay head after P2 quit = %q; want %q", after.Cell.Head, before.Cell.Head) + } +} + +func refuseHeldChatOpen(t *testing.T, cfg session.Config, writes *atomic.Int64) session.Config { + t.Helper() + _, seated, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + var held *sessionHeldElsewhere + if !errors.As(err, &held) { + t.Fatalf("P2 open = %v; want sessionHeldElsewhere from the real journal lock", err) + } + // On the unfixed product startup is asynchronous. Observe its first write + // before quitting; closeAll then joins startup and deterministically releases it. + deadline := time.Now().Add(2 * time.Second) + for writes.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + syncDrives.closeAll() + return seated +} + +func TestAChatAnotherProcessHoldsIsNeitherRenewedNorReleasedFromHere(t *testing.T) { + celltest.Available(t) + t.Setenv(home.EnvVar, t.TempDir()) + t.Setenv("HOME", t.TempDir()) + // A missing explicit binary prevents even an installed engine from being + // used by the failed opener's background warm capture. P1 needs no seals. + t.Setenv("CODEAF_FURROW", filepath.Join(t.TempDir(), "absent-engine")) + c, err := cell.CreateIn(t.TempDir(), cell.Options{Class: cell.HostBound}) + if err != nil { + t.Fatalf("create the held chat: %v", err) + } + place := session.Place{Dir: c.Root, Workspace: t.TempDir()} + cfg := session.Config{Place: place, Workspace: place.Workspace, SessionFile: place.Transcript(), Model: "stub/chat", APIKey: "stub", BaseURL: "http://127.0.0.1:1"} + journal, err := os.OpenFile(place.Transcript(), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) + if err != nil { + t.Fatalf("open P1's journal: %v", err) + } + defer journal.Close() + if err := filelock.Lock(journal, true, true); err != nil { + t.Fatalf("hold P1's journal with the session's exclusive nonblocking flock: %v", err) + } + defer filelock.Unlock(journal) + s, writes := heldChatRelay(t, c.ID) + if _, err := s.Dir.Create(t.Context(), c.ID, directory.CellInit{}); err != nil { + t.Fatalf("create P1's relay record and lease: %v", err) + } + prior := syncDrives + firstBook := &driveBook{} + syncDrives = firstBook + defer func() { syncDrives.closeAll(); syncDrives = prior }() + // P1 resumes the relay's fence and records its receipt through the real + // Drive. Its sleeper holds all renewal ticks until Close cancels them. + first := newLiveDrive(func(func()) (*syncsetup.Drive, error) { + return s.Drive(t.Context(), cellstore.Engine{}, c, syncsetup.DriveOptions{Sleep: func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + }}) + }) + firstBook.drive = map[string]*liveDrive{c.ID: first} + defer firstBook.closeAll() + if err := first.Gate(); err != nil { + t.Fatalf("P1's initial gate: %v", err) + } + writer := first.started() + if writer == nil || !writer.HoldsLease(t.Context()) { + t.Fatalf("P1 must start a real writer holding the relay lease") + } + before, err := s.Dir.Cell(t.Context(), c.ID) + if err != nil || before.Cell.Lease.Fence < 1 || before.Cell.Lease.Expires <= before.Now { + t.Fatalf("P1's initial relay lease = %+v, %v; want a live fence", before, err) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + writes.Store(0) + syncDrives = &driveBook{} + seated := refuseHeldChatOpen(t, cfg, writes) + // Drain only this failed opener's warm capture before removing its files. + // Warm preparation screens locally and does not require a sealing binary. + _ = executor.Drain(t.Context(), seated.Seat) + if got := writes.Load(); got != 0 { + t.Fatalf("P2 made %d lease writes while opening and quitting a held chat; want 0", got) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + if !writer.HoldsLease(t.Context()) { + t.Fatalf("P1 lost its lease when P2 quit") + } + if err := writer.Gate(); err != nil { + t.Fatalf("P1's next work gate = %v; want nil", err) + } + confirmed, err := writer.ConfirmWork(t.Context()) + if err != nil || !confirmed { + t.Fatalf("P1's next work confirmation = %v, %v; want true, nil", confirmed, err) + } + if d := syncDrives.renewalsOf(c.ID); d != nil && d.started() != nil { + t.Fatalf("P2 retained a started drive for a chat whose journal it never held") + } +} + +func TestAQuitAtTheMoveQuestionLeavesTheOtherWindowsChatSendable(t *testing.T) { + cfg, c := spokenDoorChat(t) + s, writes := heldChatRelay(t, c.ID) + firstBook := syncDrives + defer firstBook.closeAll() + agent, firstSeat, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + if err != nil { + t.Fatalf("open P1's seated chat: %v", err) + } + defer agent.Close() + send := func(command string) { + t.Helper() + events, err := agent.SubmitBash(t.Context(), command) + if err != nil { + t.Fatalf("P1 send %q = %v; want admitted", command, err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("P1 send %q reported %q; want no EventError", command, event.Text) + } + } + if err := executor.Drain(t.Context(), firstSeat.Seat); err != nil { + t.Fatalf("settle P1 send %q: %v", command, err) + } + } + send("! printf fixture") + deadline := time.Now().Add(15 * time.Second) + var before directory.CellView + for { + head, headErr := cellstore.Head(c) + before, err = s.Dir.Cell(t.Context(), c.ID) + _, receiptFence, receiptErr := s.SendLease(c.ID) + if headErr == nil && head != nil && err == nil && before.Cell.Head == head.Turn.ID && before.Cell.Lease.Fence > 0 && receiptErr == nil && receiptFence == before.Cell.Lease.Fence { + break + } + if time.Now().After(deadline) { + t.Fatalf("P1's spoken turn did not publish: relay = %+v, %v; local head error = %v", before, err, headErr) + } + time.Sleep(10 * time.Millisecond) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + first := firstBook.renewalsOf(c.ID) + writes.Store(0) + syncDrives = &driveBook{} + t.Cleanup(func() { syncDrives.closeAll(); syncDrives = firstBook }) + seated := refuseHeldChatOpen(t, cfg, writes) + if err := executor.Drain(t.Context(), seated.Seat); err != nil { + t.Fatalf("settle P2's unused seat: %v", err) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + send("! printf again") + turns, err := cellstore.Turns(c) + if err != nil || len(turns) == 0 { + t.Fatalf("read P1's newest sealed turn = %d turns, %v", len(turns), err) + } + // A sealed turn names its writer by the device certificate's stamp, which + // is what the drive's store seals with, not by the relay's device id. + newest := turns[len(turns)-1] + if newest.Device != s.Device.Cert.Device || newest.Fence != before.Cell.Lease.Fence { + t.Fatalf("P1's next turn = %s fence %d; want %s fence %d", newest.Device, newest.Fence, s.Device.Cert.Device, before.Cell.Lease.Fence) + } + if first == nil || first.started() == nil || !first.started().HoldsLease(t.Context()) { + t.Fatalf("P1's drive must still hold its original lease after its next send") + } +} + +func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { + celltest.Available(t) + t.Setenv(home.EnvVar, t.TempDir()) + t.Setenv(syncsetup.URLVar, "http://relay.invalid") + + fixture := func(t *testing.T) (*driveBook, *liveDrive, *atomic.Int64, func(bool)) { + t.Helper() + calls := &atomic.Int64{} + d := newAdmissionDrive(func(func()) (*syncsetup.Drive, error) { + calls.Add(1) + return nil, nil + }) + book := &driveBook{drive: map[string]*liveDrive{"chat": d}} + t.Cleanup(book.closeAll) + got, claim := book.pendingDriveOf(cell.Cell{ID: "chat"}, cellstore.Engine{}, func(error) {}) + if got != d { + t.Fatalf("the parked drive was replaced while its journal open was in flight") + } + return book, d, calls, claim + } + claimAgain := func(t *testing.T, book *driveBook, d *liveDrive) func(bool) { + t.Helper() + got, claim := book.pendingDriveOf(cell.Cell{ID: "chat"}, cellstore.Engine{}, func(error) {}) + if got != d { + t.Fatalf("a second opener must share the existing drive") + } + return claim + } + waiters := func(t *testing.T, d *liveDrive) func() { + t.Helper() + gate, store := make(chan error, 1), make(chan error, 1) + entered := make(chan struct{}, 2) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + t.Cleanup(cancel) + go func() { entered <- struct{}{}; gate <- d.Gate() }() + go func() { + entered <- struct{}{} + chosen, used, err := (lateStore{live: d, engine: cellstore.Engine{Workspace: "local"}}).readyStore(ctx) + if err == nil { + local, ok := chosen.(cellstore.Engine) + if !ok || local.Workspace != "local" || used != nil { + err = fmt.Errorf("store = %T, drive present = %v; want the local engine and no drive", chosen, used != nil) + } + } + store <- err + }() + <-entered + <-entered + select { + case err := <-gate: + t.Fatalf("parked gate returned %v before journal ownership was decided", err) + case err := <-store: + t.Fatalf("parked store returned %v before journal ownership was decided", err) + case <-time.After(20 * time.Millisecond): + } + return func() { + t.Helper() + for _, waiter := range []<-chan error{gate, store} { + select { + case err := <-waiter: + if err != nil { + t.Fatalf("resolved parked waiter = %v; want nil", err) + } + case <-ctx.Done(): + t.Fatalf("journal ownership was decided but its parked waiter did not return") + } + } + } + } + + t.Run("a successful journal open starts once and releases both waiters", func(t *testing.T) { + book, d, calls, claim := fixture(t) + ready := d.pending() + if ready == nil { + t.Fatalf("a parked drive must already have its startup channel") + } + finish := waiters(t, d) + d.try() + if got := calls.Load(); got != 0 || d.pending() != ready { + t.Fatalf("parked try called start %d times or replaced ready; want 0 and the original channel", got) + } + claim(true) + claim(false) + claim(true) + finish() + if got := calls.Load(); got != 1 { + t.Fatalf("successful claim called start %d times; want 1", got) + } + if book.renewalsOf("chat") != d { + t.Fatalf("the successful journal opener lost its drive") + } + }) + + t.Run("every failed journal open abandons without starting", func(t *testing.T) { + book, d, calls, first := fixture(t) + second := claimAgain(t, book, d) + finish := waiters(t, d) + first(false) + first(true) + if d.pending() == nil { + t.Fatalf("one failed opener abandoned a drive another opener still claims") + } + second(false) + second(true) + finish() + if got := calls.Load(); got != 0 { + t.Fatalf("failed journal opens called start %d times; want 0", got) + } + if book.renewalsOf("chat") != nil { + t.Fatalf("the book retained the abandoned drive") + } + if _, found := workAdmissions.Load(d); found { + t.Fatalf("the abandoned drive retained work admission state") + } + }) + + for _, operation := range []string{"Close", "closeAll", "take", "closeCell", "takeCell"} { + t.Run(operation+" abandons immediately and a later success cannot start", func(t *testing.T) { + book, d, calls, claim := fixture(t) + finish := waiters(t, d) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + done := make(chan error, 1) + go func() { + switch operation { + case "Close": + done <- d.Close(ctx) + case "closeAll": + book.closeAll() + done <- nil + case "take": + book.take() + done <- nil + case "closeCell": + done <- book.closeCell("chat") + case "takeCell": + book.takeCell("chat") + done <- nil + } + }() + select { + case err := <-done: + if err != nil { + t.Fatalf("closing parked drive = %v; want nil even with a canceled context", err) + } + case <-time.After(250 * time.Millisecond): + t.Fatalf("closing a parked drive waited for startup or driveCloseWithin") + } + claim(true) + d.try() + finish() + if got := calls.Load(); got != 0 { + t.Fatalf("claim after close called start %d times; want 0", got) + } + }) + } + + for _, successFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("two openers keep one drive when success comes first %v", successFirst), func(t *testing.T) { + book, d, calls, first := fixture(t) + second := claimAgain(t, book, d) + first(successFirst) + second(!successFirst) + if err := d.Gate(); err != nil { + t.Fatalf("the successful opener's gate = %v; want nil", err) + } + if got := calls.Load(); got != 1 { + t.Fatalf("two openers called start %d times; want 1", got) + } + if book.renewalsOf("chat") != d { + t.Fatalf("a failed opener removed the successful opener's drive") + } + if _, found := workAdmissions.Load(d); !found { + t.Fatalf("a failed opener removed the successful opener's work admission") + } + }) + } + + t.Run("a claim on startup already begun changes nothing", func(t *testing.T) { + book, d, calls, first := fixture(t) + first(true) + if err := d.Gate(); err != nil { + t.Fatalf("initial gate = %v; want nil", err) + } + claimAgain(t, book, d)(false) + claimAgain(t, book, d)(true) + d.mu.Lock() + closed := d.closed + d.mu.Unlock() + if got := calls.Load(); got != 1 || closed || book.renewalsOf("chat") != d { + t.Fatalf("claim on begun drive: starts = %d, closed = %v, kept = %v; want 1, false, true", got, closed, book.renewalsOf("chat") == d) + } + }) +} diff --git a/cmd/codeaf/syncdrive.go b/cmd/codeaf/syncdrive.go index c7b7abf989..f5b2d2df07 100644 --- a/cmd/codeaf/syncdrive.go +++ b/cmd/codeaf/syncdrive.go @@ -26,8 +26,9 @@ var syncDrives = &driveBook{} // same process (a resume, a switch back) goes on with the drive it had instead // of taking a second lease. type driveBook struct { - mu sync.Mutex - drive map[string]*liveDrive + mu sync.Mutex + drive map[string]*liveDrive + claims map[*liveDrive]int // notices is installed by an app assembly before any drive starts. Nil // retains the terminal's desk and its marking first-run behaviour. notices *driveNotices @@ -47,17 +48,25 @@ type driveNotices struct { // moment one is made (see liveDrive). A nil book answers nil, which is how a // door that never syncs asks. func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(error)) *liveDrive { + d, claim := b.pendingDriveOf(c, engine, report) + claim(true) + return d +} + +// pendingDriveOf reserves a drive until one opener owns the journal. +func (b *driveBook) pendingDriveOf(c cell.Cell, engine cellstore.Engine, report func(error)) (*liveDrive, func(bool)) { + noop := func(bool) {} if b == nil { - return nil + return nil, noop } if syncsetup.Resolve(home.Dir()).Off || !cell.Available() { - return nil + return nil, noop } b.mu.Lock() defer b.mu.Unlock() if d := b.drive[c.ID]; d != nil { if !d.superseded() { - return d + return d, b.claim(d, c.ID) } // Settle the displaced writer before the replacement reads its branch map // or acquires a lease, so late uploads cannot change the new drive's target. @@ -66,7 +75,7 @@ func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(er cancel() if err != nil { report(err) - return d + return d, b.claim(d, c.ID) } } // A drive side that no longer holds the chat's lease (another machine took it, or it was moved @@ -84,7 +93,36 @@ func (b *driveBook) driveOf(c cell.Cell, engine cellstore.Engine, report func(er } b.drive[c.ID] = d } - return d + return d, b.claim(d, c.ID) +} + +// claim is made under the book lock; no drive path takes that lock in return. +func (b *driveBook) claim(d *liveDrive, id string) func(bool) { + if !d.isParked() { + return func(bool) {} + } + if b.claims == nil { + b.claims = make(map[*liveDrive]int) + } + b.claims[d]++ + var once sync.Once + return func(opened bool) { + once.Do(func() { + b.mu.Lock() + defer b.mu.Unlock() + b.claims[d]-- + if opened { + d.begin() + } + if b.claims[d] <= 0 { + delete(b.claims, d) + if d.abandon() && b.drive[id] == d { + delete(b.drive, id) + workAdmissions.Delete(d) + } + } + }) + } } // startDrive opens the drive side of c. The error it answers is already @@ -193,6 +231,7 @@ func (b *driveBook) take() map[string]*liveDrive { defer b.mu.Unlock() drives := b.drive for _, d := range drives { + d.abandon() workAdmissions.Delete(d) } b.drive = nil @@ -254,6 +293,9 @@ func (b *driveBook) takeCell(id string) *liveDrive { b.mu.Lock() defer b.mu.Unlock() d := b.drive[id] + if d != nil { + d.abandon() + } workAdmissions.Delete(d) delete(b.drive, id) return d diff --git a/cmd/codeaf/syncdrive_admission.go b/cmd/codeaf/syncdrive_admission.go index 5f88b200b5..0d389d3b0d 100644 --- a/cmd/codeaf/syncdrive_admission.go +++ b/cmd/codeaf/syncdrive_admission.go @@ -39,7 +39,7 @@ type workAdmission struct { func newAdmissionDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { check := &workAdmission{} - d := newLiveDrive(func(changed func()) (*syncsetup.Drive, error) { + d := newPendingLiveDrive(func(changed func()) (*syncsetup.Drive, error) { drive, err := start(changed) if err == nil && drive != nil && drive.Gate() == nil { check.mu.Lock() diff --git a/cmd/codeaf/syncdrive_live.go b/cmd/codeaf/syncdrive_live.go index 21246e594e..b5faba5e9f 100644 --- a/cmd/codeaf/syncdrive_live.go +++ b/cmd/codeaf/syncdrive_live.go @@ -27,6 +27,7 @@ type liveDrive struct { mu sync.Mutex drive *syncsetup.Drive + parked bool // Startup awaits a journal owner. waiting bool // sync is on and only an identity, or the end of solo, is missing ready chan struct{} attempted bool @@ -40,9 +41,15 @@ type liveDrive struct { // newLiveDrive starts the drive side behind the first paint. Local startup // screening stays off the surface loop; saving and new tools wait for ownership. func newLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { - l := &liveDrive{changes: make(chan struct{}, 1)} + l := newPendingLiveDrive(start) + l.begin() + return l +} + +// newPendingLiveDrive reserves the startup boundary before journal ownership is known. +func newPendingLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { + l := &liveDrive{changes: make(chan struct{}, 1), parked: true, ready: make(chan struct{})} l.start = func() (*syncsetup.Drive, error) { return start(l.renewalChanged) } - l.try() return l } @@ -52,11 +59,52 @@ func newLiveDrive(start func(func()) (*syncsetup.Drive, error)) *liveDrive { func (l *liveDrive) try() { l.mu.Lock() defer l.mu.Unlock() - if l.closed || l.drive != nil || l.ready != nil || (l.attempted && !l.waiting) || (l.waiting && !syncsetup.MayTalk(home.Dir())) { + if l.parked || l.closed || l.drive != nil || l.ready != nil || (l.attempted && !l.waiting) || (l.waiting && !syncsetup.MayTalk(home.Dir())) { return } l.ready = make(chan struct{}) - ready := l.ready + l.launch(l.ready) +} + +// begin spends the reserved startup exactly once after the journal opens. +func (l *liveDrive) begin() { + l.mu.Lock() + defer l.mu.Unlock() + if !l.parked { + return + } + l.parked = false + l.launch(l.ready) +} + +// isParked says the startup is still waiting for a journal owner, so an opener +// of the same chat in this process shares that wait rather than starting it. +func (l *liveDrive) isParked() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.parked +} + +// abandon releases parked waiters without ever opening the drive side. +func (l *liveDrive) abandon() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.abandonLocked() +} + +func (l *liveDrive) abandonLocked() bool { + if !l.parked { + return false + } + l.parked, l.closed = false, true + close(l.ready) + l.ready = nil + return true +} + +// launch shares startup and panic cleanup between immediate and reserved starts. +// The caller holds mu, so closing cannot slip between reserving and launching. +func (l *liveDrive) launch(ready chan struct{}) { guard.Go("chat sync startup", func() { var d *syncsetup.Drive var err error @@ -207,6 +255,7 @@ func (l *liveDrive) Close(ctx context.Context) error { ready := func() <-chan struct{} { l.mu.Lock() defer l.mu.Unlock() + l.abandonLocked() l.closed = true if l.retry != nil { l.retry.Stop() diff --git a/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md b/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md new file mode 100644 index 0000000000..8619d3a0a8 --- /dev/null +++ b/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md @@ -0,0 +1,13 @@ +--- +kind: fixed +title: Opening a held chat leaves the other window's sync lease alone +pr: 0 +surface: [chat, engine] +invalidates: + - "Opening a chat used to start its sync drive before checking the journal lock, so a second process could renew the other window's fence and release it on quit. A process now starts that chat's sync drive only after opening its journal successfully." + - "Quitting at the move question or stopping a host after a refused open could delete the original writer's hold receipt. A refused journal open now leaves the original writer's lease and receipt untouched." +--- + +A terminal opened beside a chat held by the desktop app or another process can +still offer to move the conversation here. Leaving that question, quitting the +terminal, or stopping its host keeps the original window's chat sendable. From 145db19561f122d9502e6850dd9d682f836764df Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 16:56:49 -0400 Subject: [PATCH 2/4] test(chat): the parked-drive cases wait on the boundary, not on the clock The waiters are asserted parked when the goroutine dump shows them blocked inside Gate and readyStore, rather than after a 20 ms sleep, and closing a parked drive is bounded by a third of driveCloseWithin rather than 250 ms, so a loaded machine cannot fail correct code. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/heldchat_drive_test.go | 34 ++++++++++++++++++++++++++++--- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/cmd/codeaf/heldchat_drive_test.go b/cmd/codeaf/heldchat_drive_test.go index 80824ebd6d..b02acd1232 100644 --- a/cmd/codeaf/heldchat_drive_test.go +++ b/cmd/codeaf/heldchat_drive_test.go @@ -10,6 +10,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "runtime" "strings" "sync/atomic" "testing" @@ -274,6 +275,27 @@ func TestAQuitAtTheMoveQuestionLeavesTheOtherWindowsChatSendable(t *testing.T) { } } +// waitParkedIn waits until some goroutine is blocked on a channel inside fn, +// which is how a waiter on a parked drive's startup boundary looks. +func waitParkedIn(t *testing.T, fn string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + buf := make([]byte, 1<<20) + for { + n := runtime.Stack(buf, true) + for _, g := range strings.Split(string(buf[:n]), "\n\n") { + head, _, _ := strings.Cut(g, "\n") + if (strings.Contains(head, "[chan receive") || strings.Contains(head, "[select")) && strings.Contains(g, fn) { + return + } + } + if time.Now().After(deadline) { + t.Fatalf("no goroutine is waiting inside %s on the parked startup", fn) + } + time.Sleep(5 * time.Millisecond) + } +} + func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { celltest.Available(t) t.Setenv(home.EnvVar, t.TempDir()) @@ -306,7 +328,7 @@ func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { t.Helper() gate, store := make(chan error, 1), make(chan error, 1) entered := make(chan struct{}, 2) - ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) t.Cleanup(cancel) go func() { entered <- struct{}{}; gate <- d.Gate() }() go func() { @@ -322,12 +344,16 @@ func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { }() <-entered <-entered + // Both waiters must be parked on the startup boundary, not merely + // scheduled: the goroutine dump names each one blocked inside its wait. + waitParkedIn(t, "(*liveDrive).Gate") + waitParkedIn(t, "lateStore.readyStore") select { case err := <-gate: t.Fatalf("parked gate returned %v before journal ownership was decided", err) case err := <-store: t.Fatalf("parked store returned %v before journal ownership was decided", err) - case <-time.After(20 * time.Millisecond): + default: } return func() { t.Helper() @@ -419,7 +445,9 @@ func TestAParkedChatDriveWaitsForItsJournalDecision(t *testing.T) { if err != nil { t.Fatalf("closing parked drive = %v; want nil even with a canceled context", err) } - case <-time.After(250 * time.Millisecond): + case <-time.After(driveCloseWithin / 3): + // A close that waited for the parked startup would take the + // whole driveCloseWithin; a third of it is generous under load. t.Fatalf("closing a parked drive waited for startup or driveCloseWithin") } claim(true) From f19c10ba77e5f26bb0b8dbe989ef55dd34150f54 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 17:39:20 -0400 Subject: [PATCH 3/4] chat: the fresh chat beside a held one opens on its own seat v3TakeOverInstead pointed the refused launch's config at a fresh folder and opened it directly, so the fresh chat kept the held chat's seat, seals and renewals: its commands ran and sealed on the held chat's cell. It now opens through openV3Agent, the road /new takes, so the fresh chat gets its own seat and its own drive, begun once its journal opens. Tests follow the drive starting once the journal is held: - the take round trip waits for the reopened chat's new hold before the other computer reads its row, instead of racing that acquire; - C6's startup boundary does not count the device and chat listings, which the drive's startup and the app's inventory send in either order; a cell read by admission is still counted; - the held-chat lease check allows the holder's own renewal while it still refuses a release, and the parked waiters' dump grows until it is complete. Co-Authored-By: Claude Opus 5.5 --- cmd/codeaf/app_take_test.go | 21 +++ cmd/codeaf/appsync_admission_test.go | 29 +++- cmd/codeaf/chatv3.go | 7 +- cmd/codeaf/chatv3_session_test.go | 4 +- cmd/codeaf/heldchat_drive_test.go | 125 +++++++++++++++++- ...0-held-chat-drive-waits-for-its-journal.md | 7 +- 6 files changed, 177 insertions(+), 16 deletions(-) diff --git a/cmd/codeaf/app_take_test.go b/cmd/codeaf/app_take_test.go index 25a712a33f..4bf81eb4a7 100644 --- a/cmd/codeaf/app_take_test.go +++ b/cmd/codeaf/app_take_test.go @@ -322,6 +322,22 @@ func appTakeRoundTrip(t *testing.T, daemon bool, missingWorkspace ...bool) { Agent appwire.Agent `json:"agent"` }](t, a.w, "agents.create", map[string]any{"workspace": workspace}).Agent e2Turn(t, a, created.ID, "first") + heldFence := func() uint64 { + rows := machineCall[appwire.ChatsResult](t, a.w, "chats.list", struct{}{}) + for _, r := range rows.Chats { + if r.Cell == created.ID && r.Here { + return r.Seen.Fence + } + } + return 0 + } + var firstFence uint64 + if !absent { + appSyncWait(t, "A's first published fence", func() bool { + firstFence = heldFence() + return firstFence > 0 + }) + } // Start continuation from a chat saved by the real app leaving path. This // exercises a saved chat independently of background first-publication timing. e2Quit(t, a) @@ -332,6 +348,11 @@ func appTakeRoundTrip(t *testing.T, daemon bool, missingWorkspace ...bool) { } else { a = e2StartApp(t, dirA, daemon) machineCall[appwire.TakeResult](t, a.w, "agents.open", appwire.IDParams{ID: created.ID}) + // Reopening owns the journal before its background drive acquires. + // Take consent must name that new hold rather than race its arrival. + appSyncWait(t, "A's reopened chat holds a new fence", func() bool { + return heldFence() > firstFence + }) } t.Logf("created %s", string(e2JSON(created))) diff --git a/cmd/codeaf/appsync_admission_test.go b/cmd/codeaf/appsync_admission_test.go index 00aaa54f0c..31d706014d 100644 --- a/cmd/codeaf/appsync_admission_test.go +++ b/cmd/codeaf/appsync_admission_test.go @@ -50,10 +50,29 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { entered, release := make(chan struct{}), make(chan struct{}) var once sync.Once unblock := func() { once.Do(func() { close(release) }) } - var requests atomic.Int64 + var requestMu sync.Mutex + var requestPaths []string + recordRequest := func(r *http.Request) { + requestMu.Lock() + defer requestMu.Unlock() + requestPaths = append(requestPaths, r.Method+" "+r.URL.Path) + } + requests := func() []string { + requestMu.Lock() + defer requestMu.Unlock() + return append([]string(nil), requestPaths...) + } var first sync.Once srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - requests.Add(1) + // The drive's startup opens with a device listing, at the same + // moment as the app's own chat listing and watch. Every request + // is held, so a startup held at its listing cannot reach the + // chat's cell; the listings themselves are not admission's work + // and arrive in either order, so they are not counted. + listing := r.Method == http.MethodGet && (r.URL.Path == "/v1/dir/list" || r.URL.Path == "/v1/dir/watch") + if boundary != "startup" || !listing { + recordRequest(r) + } if boundary == "startup" { first.Do(func() { close(entered) }) select { @@ -120,7 +139,7 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { } }) } - before := requests.Load() + before := len(requests()) type sent struct { turn uint64 err error @@ -145,8 +164,8 @@ func TestC6AppSendDoesNotWaitForSyncStartupOrCapture(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("the admitted send never reached the real session model") } - if got := requests.Load(); got != before { - t.Fatalf("admission added relay work: before %d, after %d", before, got) + if got := requests(); len(got) != before { + t.Fatalf("admission added relay work: before %d, after %d; added %v", before, len(got), got[before:]) } unblock() releaseModel() diff --git a/cmd/codeaf/chatv3.go b/cmd/codeaf/chatv3.go index 13c113852e..730e2e553c 100644 --- a/cmd/codeaf/chatv3.go +++ b/cmd/codeaf/chatv3.go @@ -1581,11 +1581,8 @@ func v3TakeOverInstead(cfg session.Config, workspace string) (*session.Agent, se if err != nil { return nil, cfg, err } - agent, err := v3OpenSession(fresh) - if err != nil { - return nil, cfg, err - } - return agent, fresh, nil + agent, fresh, _, err := openV3Agent(fresh, workspace, v3OpenSession) + return agent, fresh, err } // sessionHeldElsewhereSentence is what a person reads when the conversation diff --git a/cmd/codeaf/chatv3_session_test.go b/cmd/codeaf/chatv3_session_test.go index c68b4364bd..01fe0541dd 100644 --- a/cmd/codeaf/chatv3_session_test.go +++ b/cmd/codeaf/chatv3_session_test.go @@ -143,7 +143,7 @@ func TestALaunchThatMeetsALockOpensAFreshOneAndCarriesTheHeldRow(t *testing.T) { } defer func() { _ = first.Close() }() - _, _, _, err = openV3Agent(cfg, workspace, v3OpenSession) + _, seated, _, err := openV3Agent(cfg, workspace, v3OpenSession) // THE FAILURE IS TYPED AND CARRIES THE PATH, because the sentence // deliberately does not contain one and the surface needs one to point at. var held *sessionHeldElsewhere @@ -159,7 +159,7 @@ func TestALaunchThatMeetsALockOpensAFreshOneAndCarriesTheHeldRow(t *testing.T) { t.Fatalf("%v no longer reads as a locked session", err) } - second, fresh, err := v3TakeOverInstead(cfg, workspace) + second, fresh, err := v3TakeOverInstead(seated, workspace) if err != nil { t.Fatalf("the second window was left with nothing: %v", err) } diff --git a/cmd/codeaf/heldchat_drive_test.go b/cmd/codeaf/heldchat_drive_test.go index b02acd1232..ef508eee0a 100644 --- a/cmd/codeaf/heldchat_drive_test.go +++ b/cmd/codeaf/heldchat_drive_test.go @@ -105,8 +105,8 @@ func checkHeldChatLease(t *testing.T, s *syncsetup.Sync, id string, before direc if after.Cell.Lease.Device != before.Cell.Lease.Device || after.Cell.Lease.Fence != before.Cell.Lease.Fence { t.Fatalf("lease after P2 quit = %s fence %d; want %s fence %d", after.Cell.Lease.Device, after.Cell.Lease.Fence, before.Cell.Lease.Device, before.Cell.Lease.Fence) } - if after.Cell.Lease.Expires == 0 || after.Cell.Lease.Expires != before.Cell.Lease.Expires { - t.Fatalf("lease expiry after P2 quit = %d; want unchanged, unreleased %d", after.Cell.Lease.Expires, before.Cell.Lease.Expires) + if after.Cell.Lease.Expires == 0 || after.Cell.Lease.Expires < before.Cell.Lease.Expires { + t.Fatalf("lease expiry after P2 quit = %d; want unreleased and at least %d", after.Cell.Lease.Expires, before.Cell.Lease.Expires) } if after.Cell.Head != before.Cell.Head { t.Fatalf("relay head after P2 quit = %q; want %q", after.Cell.Head, before.Cell.Head) @@ -275,6 +275,123 @@ func TestAQuitAtTheMoveQuestionLeavesTheOtherWindowsChatSendable(t *testing.T) { } } +func TestTheFreshChatBesideAHeldOneWorksOnItsOwnSeat(t *testing.T) { + cfg, c := spokenDoorChat(t) + s, writes := heldChatRelay(t, c.ID) + firstBook := syncDrives + defer firstBook.closeAll() + agent, firstCfg, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + if err != nil { + t.Fatalf("open P1's seated chat: %v", err) + } + defer agent.Close() + events, err := agent.SubmitBash(t.Context(), "! printf fixture") + if err != nil { + t.Fatalf("P1 send = %v; want admitted", err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("P1 send reported %q; want no EventError", event.Text) + } + } + if err := executor.Drain(t.Context(), firstCfg.Seat); err != nil { + t.Fatalf("settle P1's spoken turn: %v", err) + } + deadline := time.Now().Add(15 * time.Second) + var before directory.CellView + for { + head, headErr := cellstore.Head(c) + before, err = s.Dir.Cell(t.Context(), c.ID) + _, receiptFence, receiptErr := s.SendLease(c.ID) + if headErr == nil && head != nil && err == nil && before.Cell.Head == head.Turn.ID && before.Cell.Lease.Fence > 0 && receiptErr == nil && receiptFence == before.Cell.Lease.Fence { + break + } + if time.Now().After(deadline) { + t.Fatalf("P1's spoken turn did not publish: relay = %+v, %v; local head error = %v", before, err, headErr) + } + time.Sleep(10 * time.Millisecond) + } + heldTurns, err := cellstore.Turns(c) + if err != nil || len(heldTurns) == 0 { + t.Fatalf("P1's published turns = %d, %v; want a sealed turn", len(heldTurns), err) + } + receipt := snapshotHeldChatReceipt(t, s, c.ID) + first := firstBook.renewalsOf(c.ID) + writes.Store(0) + syncDrives = &driveBook{} + defer func() { syncDrives.closeAll(); syncDrives = firstBook }() + _, seated, _, err := openV3Agent(cfg, cfg.Place.Workspace, session.New) + var held *sessionHeldElsewhere + if !errors.As(err, &held) { + t.Fatalf("P2 open = %v; want sessionHeldElsewhere", err) + } + fresh, freshCfg, err := v3TakeOverInstead(seated, cfg.Place.Workspace) + if err != nil { + t.Fatalf("open the fresh chat beside P1: %v", err) + } + defer fresh.Close() + if freshCfg.Place.Dir == c.Root { + t.Fatal("the fresh chat shares P1's cell") + } + if err := executor.Drain(t.Context(), seated.Seat); err != nil { + t.Fatalf("settle P2's unused seat: %v", err) + } + if err := executor.Drain(t.Context(), freshCfg.Seat); err != nil { + t.Fatalf("settle the fresh chat's startup: %v", err) + } + freshCell, err := cell.OpenAt(freshCfg.Place.Dir, freshCfg.Place.ID()) + if err != nil { + t.Fatalf("open the fresh chat's cell: %v", err) + } + freshBefore, err := cellstore.Turns(freshCell) + if err != nil { + t.Fatalf("read the fresh chat before sending: %v", err) + } + events, err = fresh.SubmitBash(t.Context(), "! printf beside") + if err != nil { + t.Fatalf("fresh chat send = %v; want admitted", err) + } + for event := range events { + if event.Kind == session.EventError { + t.Fatalf("fresh chat send reported %q; want no EventError", event.Text) + } + } + if err := executor.Drain(t.Context(), freshCfg.Seat); err != nil { + t.Fatalf("settle the fresh chat's command: %v", err) + } + checkHeld := func() { + t.Helper() + turns, err := cellstore.Turns(c) + if err != nil || len(turns) != len(heldTurns) { + t.Fatalf("P1's sealed turns = %d, %v; want unchanged count %d", len(turns), err, len(heldTurns)) + } + if got, want := turns[len(turns)-1].ID, heldTurns[len(heldTurns)-1].ID; got != want { + t.Fatalf("P1's newest turn = %s; want unchanged %s", got, want) + } + if got := writes.Load(); got != 0 { + t.Fatalf("P2 made %d writes to P1's relay cell; want 0", got) + } + checkHeldChatLease(t, s, c.ID, before) + receipt.check(t) + if first == nil || first.started() == nil || !first.started().HoldsLease(t.Context()) { + t.Fatal("P1 lost its hold while P2 used or quit the fresh chat") + } + } + checkHeld() + turns, err := cellstore.Turns(freshCell) + if err != nil || len(turns) <= len(freshBefore) { + t.Fatalf("fresh chat's sealed turns = %d, %v; want a new turn beyond %d", len(turns), err, len(freshBefore)) + } + if d := syncDrives.renewalsOf(freshCell.ID); d == nil || d.started() == nil { + t.Fatal("P2 has no started drive for the fresh chat's own cell") + } + if d := syncDrives.renewalsOf(c.ID); d != nil && d.started() != nil { + t.Fatal("P2 started a drive for P1's held chat") + } + syncDrives.closeAll() + checkHeld() +} + // waitParkedIn waits until some goroutine is blocked on a channel inside fn, // which is how a waiter on a parked drive's startup boundary looks. func waitParkedIn(t *testing.T, fn string) { @@ -283,6 +400,10 @@ func waitParkedIn(t *testing.T, fn string) { buf := make([]byte, 1<<20) for { n := runtime.Stack(buf, true) + for n == len(buf) { + buf = make([]byte, len(buf)*2) + n = runtime.Stack(buf, true) + } for _, g := range strings.Split(string(buf[:n]), "\n\n") { head, _, _ := strings.Cut(g, "\n") if (strings.Contains(head, "[chan receive") || strings.Contains(head, "[select")) && strings.Contains(g, fn) { diff --git a/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md b/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md index 8619d3a0a8..4ced8d68fc 100644 --- a/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md +++ b/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md @@ -4,10 +4,13 @@ title: Opening a held chat leaves the other window's sync lease alone pr: 0 surface: [chat, engine] invalidates: - - "Opening a chat used to start its sync drive before checking the journal lock, so a second process could renew the other window's fence and release it on quit. A process now starts that chat's sync drive only after opening its journal successfully." - - "Quitting at the move question or stopping a host after a refused open could delete the original writer's hold receipt. A refused journal open now leaves the original writer's lease and receipt untouched." + - "Opening a chat used to start its sync drive before checking the journal lock. A process refused the journal no longer starts, renews, releases or forgets that chat's sync hold." + - "The fresh chat offered beside a held chat used to inherit the held chat's executor and seals. It now opens on its own seat with its own sync drive." --- A terminal opened beside a chat held by the desktop app or another process can still offer to move the conversation here. Leaving that question, quitting the terminal, or stopping its host keeps the original window's chat sendable. +The fresh chat offered beside it runs commands and saves turns in its own cell. +This does not change the overlap when a drive already started in this process +outlives its released journal and a later open is refused. From 65389d24c498ddbc8ba461cf3e4edc750f06954d Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 17:53:22 -0400 Subject: [PATCH 4/4] docs(changes): number the held-chat drive entry by its pull request Co-Authored-By: Claude Opus 5.5 --- ...journal.md => 2046-held-chat-drive-waits-for-its-journal.md} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename docs/changes/unreleased/{0-held-chat-drive-waits-for-its-journal.md => 2046-held-chat-drive-waits-for-its-journal.md} (99%) diff --git a/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md b/docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md similarity index 99% rename from docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md rename to docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md index 4ced8d68fc..ea23b855ae 100644 --- a/docs/changes/unreleased/0-held-chat-drive-waits-for-its-journal.md +++ b/docs/changes/unreleased/2046-held-chat-drive-waits-for-its-journal.md @@ -1,7 +1,7 @@ --- kind: fixed title: Opening a held chat leaves the other window's sync lease alone -pr: 0 +pr: 2046 surface: [chat, engine] invalidates: - "Opening a chat used to start its sync drive before checking the journal lock. A process refused the journal no longer starts, renews, releases or forgets that chat's sync hold."