fix(swarm): stop original member before handoff

internal/swarm/coordinator.go

@@ -57,6 +57,7 @@ type Task struct {
 	SessionID string
 	CreatedAt time.Time
 	UpdatedAt time.Time
+	handoff   bool // internal claim; status stays non-terminal until the source exits
 }
 
 // agentColors palette for stable per-agent coloring in the TUI/status. Provider
@@ -66,11 +67,12 @@ var agentColors = []string{"cyan", "magenta", "green", "yellow", "blue", "red"}
 // Coordinator is the in-memory task registry + team color assigner shared by an
 // orchestrator and its members. It is safe for concurrent use.
 type Coordinator struct {
-	mu         sync.RWMutex
-	tasks      map[string]*Task
-	colors     map[string]string // agentID -> color
-	colorIndex int
-	now        func() time.Time // injectable clock for tests
+	mu                  sync.RWMutex
+	tasks               map[string]*Task
+	handoffReservations map[string]string // successor task id -> claimed source task id
+	colors              map[string]string // agentID -> color
+	colorIndex          int
+	now                 func() time.Time // injectable clock for tests
 	// changed is closed (and replaced) on every task state change so WaitSettled
 	// can block for a transition without polling. Always non-nil after construction.
 	changed chan struct{}
@@ -79,10 +81,11 @@ type Coordinator struct {
 // NewCoordinator returns an empty coordinator using the wall clock.
 func NewCoordinator() *Coordinator {
 	return &Coordinator{
-		tasks:   map[string]*Task{},
-		colors:  map[string]string{},
-		now:     time.Now,
-		changed: make(chan struct{}),
+		tasks:               map[string]*Task{},
+		handoffReservations: map[string]string{},
+		colors:              map[string]string{},
+		now:                 time.Now,
+		changed:             make(chan struct{}),
 	}
 }
 
@@ -112,6 +115,9 @@ func (c *Coordinator) Register(id, agentID, team, description string) (Task, err
 	if _, ok := c.tasks[id]; ok {
 		return Task{}, fmt.Errorf("%w: %s", ErrTaskExists, id)
 	}
+	if _, reserved := c.handoffReservations[id]; reserved {
+		return Task{}, fmt.Errorf("%w: %s is reserved for handoff", ErrTaskExists, id)
+	}
 	now := c.now()
 	t := &Task{
 		ID:          id,
@@ -144,12 +150,132 @@ func (c *Coordinator) SetStatus(id string, status TaskStatus) error {
 	if t.Status.terminal() && t.Status != status {
 		return fmt.Errorf("swarm: task %s is %s (terminal); cannot move to %s", id, t.Status, status)
 	}
+	if t.handoff {
+		return fmt.Errorf("swarm: task %s is being handed off; cannot move to %s", id, status)
+	}
 	t.Status = status
 	t.UpdatedAt = c.now()
 	c.notifyChangeLocked()
 	return nil
 }
 
+// BeginHandoff atomically claims a non-terminal task for one handoff caller
+// without reporting it terminal. The visible status remains pending/running
+// until CommitHandoff, so collectors cannot observe settled state while the
+// source member is still unwinding.
+func (c *Coordinator) BeginHandoff(id string) (Task, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	t, ok := c.tasks[id]
+	if !ok {
+		return Task{}, fmt.Errorf("%w: %s", ErrUnknownTask, id)
+	}
+	if t.Status.terminal() {
+		return Task{}, fmt.Errorf("swarm: task %s already %s; cannot hand off", id, t.Status)
+	}
+	if t.handoff {
+		return Task{}, fmt.Errorf("swarm: task %s is already being handed off", id)
+	}
+	t.handoff = true
+	return *t, nil
+}
+
+// AbortHandoff releases a claim and any successor reservation when handoff
+// preparation fails or a stopped source does not reach its completion barrier.
+func (c *Coordinator) AbortHandoff(id string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if t, ok := c.tasks[id]; ok && t.handoff && !t.Status.terminal() {
+		t.handoff = false
+	}
+	for successorID, sourceID := range c.handoffReservations {
+		if sourceID == id {
+			delete(c.handoffReservations, successorID)
+		}
+	}
+}
+
+// ReserveHandoffSuccessor claims successorID for sourceID before any mailbox or
+// cancellation side effect. Register rejects reserved IDs, so another Swarm
+// sharing this Coordinator cannot steal the id between note delivery and the
+// atomic handoff commit.
+func (c *Coordinator) ReserveHandoffSuccessor(sourceID, successorID string) error {
+	if successorID == "" {
+		return errors.New("swarm: successor task id is required")
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	source, ok := c.tasks[sourceID]
+	if !ok {
+		return fmt.Errorf("%w: %s", ErrUnknownTask, sourceID)
+	}
+	if !source.handoff || source.Status.terminal() {
+		return fmt.Errorf("swarm: task %s has no active handoff claim", sourceID)
+	}
+	if _, exists := c.tasks[successorID]; exists {
+		return fmt.Errorf("%w: %s", ErrTaskExists, successorID)
+	}
+	if _, reserved := c.handoffReservations[successorID]; reserved {
+		return fmt.Errorf("%w: %s is reserved for handoff", ErrTaskExists, successorID)
+	}
+	if c.handoffReservations == nil {
+		c.handoffReservations = map[string]string{}
+	}
+	c.handoffReservations[successorID] = sourceID
+	return nil
+}
+
+// CommitHandoff atomically registers the successor and publishes the source's
+// handed-off terminal state. The successor is inserted first while c.mu keeps
+// the intermediate state invisible, so an ID collision cannot retire the
+// source and orphan adoption cannot observe a successor without its source
+// transition committed.
+func (c *Coordinator) CommitHandoff(sourceID, successorID, successorAgentID, team, description string) (Task, error) {
+	if successorID == "" {
+		return Task{}, errors.New("swarm: successor task id is required")
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	source, ok := c.tasks[sourceID]
+	if !ok {
+		return Task{}, fmt.Errorf("%w: %s", ErrUnknownTask, sourceID)
+	}
+	if !source.handoff {
+		return Task{}, fmt.Errorf("swarm: task %s has no handoff in progress", sourceID)
+	}
+	if source.Status.terminal() {
+		return Task{}, fmt.Errorf("swarm: task %s already %s", sourceID, source.Status)
+	}
+	if reservedFor, ok := c.handoffReservations[successorID]; !ok || reservedFor != sourceID {
+		return Task{}, fmt.Errorf("swarm: successor task %s is not reserved for handoff from %s", successorID, sourceID)
+	}
+	if _, exists := c.tasks[successorID]; exists {
+		return Task{}, fmt.Errorf("%w: %s", ErrTaskExists, successorID)
+	}
+
+	now := c.now()
+	successor := &Task{
+		ID:          successorID,
+		AgentID:     successorAgentID,
+		Team:        team,
+		Description: description,
+		Status:      StatusPending,
+		CreatedAt:   now,
+		UpdatedAt:   now,
+	}
+	// Registration precedes retirement under the same lock. No reader can see
+	// either half until both ownership records are valid.
+	c.tasks[successorID] = successor
+	delete(c.handoffReservations, successorID)
+	c.assignColorLocked(successorAgentID)
+	source.handoff = false
+	source.Status = StatusHandedOff
+	source.UpdatedAt = now
+	c.notifyChangeLocked()
+	return *successor, nil
+}
+
 // Complete marks a task done with its result.
 func (c *Coordinator) Complete(id, result string) error {
 	return c.finish(id, StatusDone, result, "", "")
@@ -182,6 +308,9 @@ func (c *Coordinator) finish(id string, status TaskStatus, result, errMsg, sessi
 	if t.Status.terminal() {
 		return fmt.Errorf("swarm: task %s already %s", id, t.Status)
 	}
+	if t.handoff {
+		return fmt.Errorf("swarm: task %s is being handed off", id)
+	}
 	t.Status = status
 	t.Result = result
 	t.Err = errMsg
@@ -205,6 +334,9 @@ func (c *Coordinator) Reassign(id, newAgentID string) error {
 	if t.Status.terminal() {
 		return fmt.Errorf("swarm: task %s already %s; cannot reassign", id, t.Status)
 	}
+	if t.handoff {
+		return fmt.Errorf("swarm: task %s is being handed off; cannot reassign", id)
+	}
 	t.AgentID = newAgentID
 	t.Status = StatusPending
 	t.UpdatedAt = c.now()

internal/swarm/coordinator_test.go

@@ -103,6 +103,87 @@ func TestCoordinatorReassign(t *testing.T) {
 	}
 }
 
+func TestCoordinatorHandoffClaimBlocksCompetingTransitions(t *testing.T) {
+	c := NewCoordinator()
+	_, _ = c.Register("t1", "a1", "team", "desc")
+	_ = c.SetStatus("t1", StatusRunning)
+	if _, err := c.BeginHandoff("t1"); err != nil {
+		t.Fatalf("BeginHandoff: %v", err)
+	}
+	if task, _ := c.Get("t1"); task.Status != StatusRunning {
+		t.Fatalf("status during handoff = %v, want running until source exits", task.Status)
+	}
+	if _, err := c.BeginHandoff("t1"); err == nil {
+		t.Fatal("a second handoff claim must fail")
+	}
+	if err := c.Complete("t1", "late result"); err == nil {
+		t.Fatal("member completion must not win after handoff is claimed")
+	}
+	if err := c.Reassign("t1", "a2"); err == nil {
+		t.Fatal("orphan adoption must not race a claimed handoff")
+	}
+	c.AbortHandoff("t1")
+}
+
+func TestCoordinatorAbortHandoffRestoresNormalCompletion(t *testing.T) {
+	c := NewCoordinator()
+	_, _ = c.Register("t1", "a1", "team", "desc")
+	if _, err := c.BeginHandoff("t1"); err != nil {
+		t.Fatalf("BeginHandoff: %v", err)
+	}
+	c.AbortHandoff("t1")
+	if err := c.Complete("t1", "done"); err != nil {
+		t.Fatalf("Complete after AbortHandoff: %v", err)
+	}
+}
+
+func TestCoordinatorHandoffReservationBlocksRegistrationUntilAbort(t *testing.T) {
+	c := NewCoordinator()
+	_, _ = c.Register("source", "a1", "team", "desc")
+	if _, err := c.BeginHandoff("source"); err != nil {
+		t.Fatalf("BeginHandoff: %v", err)
+	}
+	if err := c.ReserveHandoffSuccessor("source", "successor"); err != nil {
+		t.Fatalf("ReserveHandoffSuccessor: %v", err)
+	}
+	if _, err := c.Register("successor", "other", "team", "collision"); !errors.Is(err, ErrTaskExists) {
+		t.Fatalf("Register reserved id error = %v, want ErrTaskExists", err)
+	}
+	c.AbortHandoff("source")
+	if _, err := c.Register("successor", "other", "team", "available again"); err != nil {
+		t.Fatalf("reservation was not released by AbortHandoff: %v", err)
+	}
+	if err := c.Complete("source", "source continued"); err != nil {
+		t.Fatalf("source claim was not released by AbortHandoff: %v", err)
+	}
+}
+
+func TestCoordinatorCommitHandoffPublishesBothSides(t *testing.T) {
+	c := NewCoordinator()
+	_, _ = c.Register("source", "a1", "team", "desc")
+	_ = c.SetStatus("source", StatusRunning)
+	if _, err := c.BeginHandoff("source"); err != nil {
+		t.Fatalf("BeginHandoff: %v", err)
+	}
+	if err := c.ReserveHandoffSuccessor("source", "successor"); err != nil {
+		t.Fatalf("ReserveHandoffSuccessor: %v", err)
+	}
+	successor, err := c.CommitHandoff("source", "successor", "a2", "team", "continued")
+	if err != nil {
+		t.Fatalf("CommitHandoff: %v", err)
+	}
+	if successor.Status != StatusPending || successor.AgentID != "a2" {
+		t.Fatalf("successor = %+v, want pending ownership by a2", successor)
+	}
+	source, _ := c.Get("source")
+	if source.Status != StatusHandedOff {
+		t.Fatalf("source status = %s, want handed-off", source.Status)
+	}
+	if _, err := c.Register("successor", "other", "team", "duplicate"); !errors.Is(err, ErrTaskExists) {
+		t.Fatalf("committed successor was not registered: %v", err)
+	}
+}
+
 func TestCoordinatorColorStability(t *testing.T) {
 	c := NewCoordinator()
 	first := c.Color("a1")