Skip to content

Commit 3d3e90e

Browse files
AbirAbbasclaude
andcommitted
syncsetup: moves, uploads and displacement include calls held for a pause
Publishing a local chat, reacquiring an expired hold and choosing the head to keep on displacement now settle held calls first, so another computer never gets a head that misses finished work. The ordinary per-command ownership check only joins a running capture. After the hold is lost, finished calls may still seal locally so displacement keeps them in the branch; new tools and bookkeeping-only seals stay refused. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent aff0b19 commit 3d3e90e

6 files changed

Lines changed: 155 additions & 13 deletions

File tree

‎internal/cellsync/batch.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,10 @@ type Batcher struct {
4747
OnLost func()
4848
OnError func(error) // a refusal (see RefusalOf) the person should hear of, once per episode
4949
Now func() time.Time // default time.Now; tests inject a fake clock
50+
// BeforeResolve settles finished calls before displacement chooses the head
51+
// to keep. A failed save must leave resolution pending rather than branch
52+
// an older head and announce that all work was kept.
53+
BeforeResolve func(context.Context) error
5054
// OnRenewalChange wakes local readers when renewal health or ownership changes.
5155
OnRenewalChange func()
5256
// Liveness, when set, is a watch socket that can vouch for the lease, so
@@ -561,6 +565,11 @@ func (b *Batcher) lost(ctx context.Context, seen Driving) {
561565
// resolve turns the orphaned turns into a branch, or, when there are none,
562566
// just tells the caller the chat is now a viewer. The caller holds flushMu.
563567
func (b *Batcher) resolve(ctx context.Context) error {
568+
if b.BeforeResolve != nil {
569+
if err := b.BeforeResolve(ctx); err != nil {
570+
return err
571+
}
572+
}
564573
head, n := b.newest()
565574
by := b.holder(ctx)
566575
if local, ok := b.Publisher.Engine.(interface {
Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package cellsync
2+
3+
import (
4+
"context"
5+
"errors"
6+
"testing"
7+
)
8+
9+
func TestSavePausesDisplacementSettlesBeforeChoosingItsHead(t *testing.T) {
10+
r := newRig(t)
11+
b := r.batcher()
12+
r.publishFirst(map[string]string{"a": "shared"})
13+
r.takeOver()
14+
var kept string
15+
b.BeforeResolve = func(context.Context) error {
16+
kept = r.seal(map[string]string{"a": "finished but held"})
17+
note(b, kept)
18+
return nil
19+
}
20+
b.Displaced(t.Context())
21+
id := r.branch.Resolve(cellID)
22+
if kept == "" || id == cellID || r.head(id).Head != kept || r.head(id).OrphanTurns != 1 {
23+
t.Fatalf("displacement chose the head before settling: kept=%q branch=%q", kept, id)
24+
}
25+
}
26+
27+
func TestSavePausesDisplacementDoesNotAcknowledgeAFailedSettle(t *testing.T) {
28+
r := newRig(t)
29+
b := r.batcher()
30+
r.publishFirst(map[string]string{"a": "shared"})
31+
note(b, r.seal(map[string]string{"a": "older pending head"}))
32+
r.takeOver()
33+
fault := errors.New("latest calls could not be saved")
34+
b.BeforeResolve = func(context.Context) error { return fault }
35+
announced := false
36+
b.OnSuperseded = func(Superseded) { announced = true }
37+
b.Displaced(t.Context())
38+
if !errors.Is(b.lastFlushErr, fault) || announced || r.branch.Resolve(cellID) != cellID || b.Pending() != 1 {
39+
t.Fatalf("failed settle acknowledged older work: err=%v announced=%v pending=%d", b.lastFlushErr, announced, b.Pending())
40+
}
41+
}

‎internal/syncsetup/confirm_work.go‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,11 @@ import (
88
"github.com/Agent-Field/codeaf/internal/directory"
99
)
1010

11-
// ConfirmWork checks only this chat's directory record. Unlike CheckLease it
12-
// never waits for a capture before reading ownership. The caller bounds ctx
13-
// and decides how to degrade a failed read. True confirms the current fence;
14-
// false means either a viewer or a drive that cannot confirm ownership yet.
11+
// ConfirmWork checks only this chat's directory record without waiting for
12+
// capture before reading ownership. Reacquisition captures closed steps before
13+
// comparing heads. The caller bounds ctx and decides how to degrade a failed
14+
// read. True confirms the current fence; false means either a viewer or a drive
15+
// that cannot confirm ownership yet.
1516
func (d *Drive) ConfirmWork(ctx context.Context) (bool, error) {
1617
held := d.sendLeaseSnapshot()
1718
if d.batcher == nil {
@@ -40,6 +41,9 @@ func (d *Drive) ConfirmWork(ctx context.Context) (bool, error) {
4041
if lease.Expires <= v.Now {
4142
d.availableAfter(held, lease)
4243
// Never resume an older tree over work saved by another computer.
44+
if err := cellstore.Captured(ctx, d.cell); err != nil {
45+
return false, err
46+
}
4347
head, err := cellstore.Head(d.cell)
4448
if err != nil || head == nil || head.Turn.ID != v.Cell.Head {
4549
return false, err

‎internal/syncsetup/copy.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -285,6 +285,9 @@ func (s *Sync) publishLocalChat(ctx context.Context, root, id string) (copyAttem
285285
if err != nil {
286286
return copyAttempt{}, err
287287
}
288+
if err := cellstore.Await(ctx, c); err != nil {
289+
return copyAttempt{}, err
290+
}
288291
head, err := cellstore.Head(c)
289292
if err != nil {
290293
return copyAttempt{}, err

‎internal/syncsetup/drive.go‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -382,12 +382,15 @@ func (s *Sync) batcher(eng cellstore.Engine, c cell.Cell, drv cellsync.Driving,
382382
Dir: s.Dir, Publisher: pub, Map: branches,
383383
NewID: func() (string, error) { return cell.NewID(time.Now()) },
384384
},
385-
Driving: &drv,
386-
Interval: s.Interval,
387-
Info: d.info,
388-
OnFlush: d.flushed(),
389-
OnSuperseded: d.becomeViewer,
390-
OnLost: d.loseHold,
385+
Driving: &drv,
386+
Interval: s.Interval,
387+
Info: d.info,
388+
OnFlush: d.flushed(),
389+
OnSuperseded: d.becomeViewer,
390+
OnLost: d.loseHold,
391+
BeforeResolve: func(ctx context.Context) error {
392+
return cellstore.Captured(ctx, c)
393+
},
391394
OnError: d.refusal,
392395
Sleep: d.opt.Sleep,
393396
Liveness: s.Liveness(),
@@ -593,7 +596,7 @@ func (d *Drive) HoldsLease(ctx context.Context) bool {
593596
// heartbeat or directory change has reached this process since the move.
594597
func (d *Drive) CheckLease(ctx context.Context) error {
595598
held := d.sendLeaseSnapshot()
596-
if err := cellstore.Captured(ctx, d.cell); err != nil {
599+
if err := cellstore.WaitCapture(ctx, d.cell); err != nil {
597600
return err
598601
}
599602
if d.batcher == nil {
@@ -624,7 +627,7 @@ func (d *Drive) CheckLease(ctx context.Context) error {
624627

625628
// retireForAnotherHolder retires this writer's hold when CheckLease finds
626629
// another computer holding the chat. It retires the hold at once so no new
627-
// seal starts, waits for any seal already past Gate, and lets the batcher keep
630+
// tool starts, waits for any seal already past Gate, and lets the batcher keep
628631
// unsent turns as a branch before the window names the new holder (#2091).
629632
// The send receipt goes before the
630633
// batcher marks itself stale, because that mark already wakes renewal readers
@@ -742,7 +745,13 @@ func (s stamping) Seal(ctx context.Context, c cell.Cell, info cellstore.TurnInfo
742745
d := s.drive
743746
d.sealMu.Lock()
744747
defer d.sealMu.Unlock()
745-
if d.Gate() != nil {
748+
d.mu.Lock()
749+
stopped := d.stopped
750+
d.mu.Unlock()
751+
// Finished calls still belong to this computer's work after its hold is
752+
// lost. Seal them locally so displacement can keep them in the branch;
753+
// the gate continues to refuse new tools and bookkeeping-only saves.
754+
if stopped || d.Gate() != nil && len(info.Calls) == 0 {
746755
return cellstore.Sealed{}, cellstore.ErrReadOnly
747756
}
748757
if d.sync.isRemoved() {
Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
package syncsetup
2+
3+
import (
4+
"context"
5+
"os"
6+
"path/filepath"
7+
"testing"
8+
"time"
9+
10+
"github.com/Agent-Field/codeaf/internal/cellstore"
11+
"github.com/Agent-Field/codeaf/internal/directory"
12+
"github.com/Agent-Field/codeaf/internal/executor"
13+
)
14+
15+
type pauseStepSeat struct {
16+
executor.Stance
17+
rec *cellstore.Recorder
18+
}
19+
20+
func (s pauseStepSeat) Settle(ctx context.Context) { s.rec.Settle(ctx) }
21+
22+
func TestSavePausesOwnershipLossKeepsHeldCallsInTheBranch(t *testing.T) {
23+
r := newDriveRig(t)
24+
ctx := t.Context()
25+
appendTo(t, transcriptOf(r.cell), `{"type":"message","role":"user","content":"start this chat"}`+"\n")
26+
first, err := r.engine.Seal(ctx, r.cell, cellstore.TurnInfo{})
27+
mustNil(t, err)
28+
_, err = r.a.Dir.Create(ctx, r.cell.ID, directory.CellInit{Head: first.Turn.ID})
29+
mustNil(t, err)
30+
drive, err := r.a.Drive(ctx, r.engine, r.cell, DriveOptions{})
31+
mustNil(t, err)
32+
t.Cleanup(func() { mustNil(t, drive.Close(context.Background())) })
33+
rec, err := cellstore.NewRecorder(executor.Local{}, drive.Store(r.engine), r.cell, r.engine.WALPath(r.cell), cellstore.Options{Async: true, Now: func() time.Time { return time.Unix(100, 0) }})
34+
mustNil(t, err)
35+
mustNil(t, rec.Around(ctx, executor.Call{Tool: "bash", Args: []byte("held call"), Deferred: true}, executor.EffectExternal, cellstore.AgentRun, func() ([]byte, bool) {
36+
mustNil(t, os.WriteFile(filepath.Join(r.work, "held.txt"), []byte("held work"), 0600))
37+
return []byte("held output"), false
38+
}))
39+
appendTo(t, transcriptOf(r.cell), `{"type":"message","role":"tool","content":"held output"}`+"\n")
40+
executor.SettleStep(ctx, pauseStepSeat{rec: rec})
41+
before, err := cellstore.Head(r.cell)
42+
mustNil(t, err)
43+
if before == nil || before.Turn.ID != first.Turn.ID {
44+
t.Fatal("held call was captured before ownership loss")
45+
}
46+
view, err := r.b.Dir.Cell(ctx, r.cell.ID)
47+
mustNil(t, err)
48+
_, err = r.b.Dir.Acquire(ctx, r.cell.ID, directory.AcquireOpts{Force: true, Expect: &directory.LeaseExpectation{Device: view.Cell.Lease.Device, Fence: view.Cell.Lease.Fence}})
49+
mustNil(t, err)
50+
mustNil(t, drive.CheckLease(ctx))
51+
branches, err := branchMapOf(r.homeA)
52+
mustNil(t, err)
53+
id := branches.Resolve(r.cell.ID)
54+
if id == r.cell.ID {
55+
t.Fatal("ownership loss did not preserve the held work as a branch")
56+
}
57+
kept, err := r.a.Dir.Cell(ctx, id)
58+
mustNil(t, err)
59+
head, err := cellstore.Head(r.cell)
60+
mustNil(t, err)
61+
if head == nil || kept.Cell.Head != head.Turn.ID || len(head.Receipt.Calls) != 1 || head.Receipt.Calls[0].Tool != "bash" || drive.Gate() == nil {
62+
t.Fatalf("branch did not carry the held call or retire its writer: head=%+v branch=%+v", head, kept.Cell)
63+
}
64+
_, recovery, err := cellstore.OpenWAL(r.engine.WALPath(r.cell))
65+
mustNil(t, err)
66+
if len(recovery.Completed) != 0 {
67+
t.Fatal("preserved call remained unsealed in the WAL")
68+
}
69+
mustNil(t, os.Remove(filepath.Join(r.work, "held.txt")))
70+
mustNil(t, r.engine.Restore(ctx, r.cell, kept.Cell.Head, []string{"held.txt"}))
71+
content, err := os.ReadFile(filepath.Join(r.work, "held.txt"))
72+
mustNil(t, err)
73+
if string(content) != "held work" {
74+
t.Fatalf("kept head did not restore the held call's file: %q", content)
75+
}
76+
}

0 commit comments

Comments
 (0)