app: bound host joins and release drains

cmd/codeaf/app.go

@@ -43,6 +43,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/Agent-Field/codeaf/internal/appwire"
@@ -52,6 +53,7 @@ import (
 	"github.com/Agent-Field/codeaf/internal/effort"
 	"github.com/Agent-Field/codeaf/internal/enginehost"
 	"github.com/Agent-Field/codeaf/internal/executor"
+	"github.com/Agent-Field/codeaf/internal/filelock"
 	"github.com/Agent-Field/codeaf/internal/guard"
 	"github.com/Agent-Field/codeaf/internal/home"
 	"github.com/Agent-Field/codeaf/internal/leave"
@@ -317,6 +319,9 @@ type appOpener func(appOpenAsk) (*appConn, error)
 // a conversation.
 const appHostDialTimeout = 500 * time.Millisecond
 
+// appHostJoinWait is one budget for discovering and joining an existing host.
+var appHostJoinWait = 5 * time.Second
+
 // appOpenInProcess opens a conversation for the app. A saved conversation that
 // a window's host already holds is JOINED there, as a second view of the same
 // conversation, because opening it here as well would be a second writer on its
@@ -412,35 +417,105 @@ func appJoinExisting(ask appOpenAsk, launch *remote.LaunchShape) (*appConn, erro
 		if agent, ok := liveStageFor(ask.Transcript); ok {
 			return appLendLive(agent, ask, hello)
 		}
-		if socket, err := enginehost.DialStartingHost(ask.Workspace); err == nil {
-			client, err := remote.Dial(socket, "", hello)
+		deadline := time.Now().Add(appHostJoinWait)
+		// Keep the workspace host's connect cap as well as the overall deadline.
+		dialer := net.Dialer{Timeout: 2 * time.Second, Deadline: deadline}
+		join := func(socket net.Conn, err error) *appConn {
+			if socket == nil {
+				return nil
+			}
+			if err == nil {
+				err = socket.SetDeadline(deadline)
+			}
 			if err == nil {
-				return appConnOf(client, func() { _ = client.Close() }, false), nil
+				var client *remote.Client
+				client, err = remote.Dial(socket, "", hello)
+				if err == nil {
+					if err = socket.SetDeadline(time.Time{}); err == nil {
+						return appConnOf(client, func() { _ = client.Close() }, false)
+					}
+					_ = client.Close()
+				}
 			}
 			_ = socket.Close()
+			return nil
+		}
+		expired := func() bool { return !time.Now().Before(deadline) }
+		unreachable := func() error {
+			return appwire.Errorf(appwire.CodeUnreachable, "This chat didn’t answer from the other codeaf window. Try again, or continue there.")
+		}
+		if conn := join(appDialStartingHost(ask.Workspace, &dialer)); conn != nil {
+			return conn, nil
+		}
+		if expired() {
+			return nil, unreachable()
 		}
 		// HELD, BUT NOT BY THIS WORKSPACE'S HOST: a crew member a manager
 		// started lives in its manager's host. Every host is asked, and the
 		// one holding it answers; opening it here as well would be refused by
 		// its journal's lock, and is never tried while somebody holds it.
 		if session.InUse(ask.Transcript) {
+			dialer.Timeout = appHostDialTimeout
 			for _, path := range enginehost.Sockets() {
-				socket, err := net.DialTimeout("unix", path, appHostDialTimeout)
-				if err != nil {
-					continue
+				if expired() {
+					return nil, unreachable()
 				}
-				client, err := remote.Dial(socket, "", hello)
-				if err == nil {
-					return appConnOf(client, func() { _ = client.Close() }, false), nil
+				if conn := join(dialer.Dial("unix", path)); conn != nil {
+					return conn, nil
 				}
-				_ = socket.Close()
+			}
+			if expired() {
+				return nil, unreachable()
 			}
 			return nil, appwire.Errorf(appwire.CodeConflict, "%s", appWindowConflict)
 		}
 	}
 	return nil, nil
 }
 
+// appDialStartingHost keeps the host's brief remove-to-listen grace while
+// bounding every connect and retry by the app's remaining join budget.
+func appDialStartingHost(workspace string, dialer *net.Dialer) (net.Conn, error) {
+	path, err := enginehost.SocketPath(workspace)
+	if err != nil {
+		return nil, err
+	}
+	const pause = 10 * time.Millisecond
+	until := time.Now().Add(250 * time.Millisecond)
+	if dialer.Deadline.Before(until) {
+		until = dialer.Deadline
+	}
+	for {
+		conn, err := dialer.Dial("unix", path)
+		if err == nil {
+			return conn, nil
+		}
+		transient := errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, os.ErrNotExist)
+		if !transient || !time.Now().Before(until) {
+			return nil, err
+		}
+		lockPath, lockErr := enginehost.LockPath(workspace)
+		if lockErr != nil {
+			return nil, err
+		}
+		file, lockErr := os.OpenFile(lockPath, os.O_RDWR, 0o600)
+		if lockErr != nil {
+			return nil, err
+		}
+		lockErr = filelock.Lock(file, true, true)
+		if lockErr == nil {
+			_ = filelock.Unlock(file)
+		}
+		_ = file.Close()
+		if !filelock.IsBusy(lockErr) {
+			return nil, err
+		}
+		if left := time.Until(until); left > 0 {
+			time.Sleep(min(pause, left))
+		}
+	}
+}
+
 func appConnOf(client *remote.Client, release func(), assembled bool) *appConn {
 	welcome := client.Welcome()
 	users := 0
@@ -1622,6 +1697,10 @@ func (b *appBackend) ensureLocked(a *appAgent) (appwire.Agent, bool, error) {
 	conn, err := b.open(ask)
 	b.locked(func() { a.opening = wasOpening })
 	if err != nil {
+		var wire *appwire.Error
+		if errors.As(err, &wire) && wire.Code == appwire.CodeUnreachable {
+			return appwire.Agent{}, false, wire
+		}
 		// Loopback errors may have crossed a codec and lost their Go type.
 		// The journal lock still proves an ownership conflict in that case.
 		if errors.Is(err, session.ErrSessionLocked) || session.InUse(transcript) {
@@ -3049,7 +3128,10 @@ func (b *appBackend) finish(live *appLive) {
 		live.stopPasses()
 	}
 	live.conn.Release()
-	wait := time.After(appReleaseWait)
+	// A closed channel lets every producer observe the same expired deadline.
+	wait := make(chan struct{})
+	stop := b.after(appReleaseWait, func() { close(wait) })
+	defer stop()
 	if live.pumpDone != nil {
 		select {
 		case <-live.pumpDone:

cmd/codeaf/appfinish_test.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Agent-Field/codeaf/internal/appwire"
+)
+
+// Contract 5: expiry while the pump is stuck must cover later stuck producers
+// too, and must still close and flush the outbox.
+func TestAppFinishSharesOneDeadline(t *testing.T) {
+	for _, lane := range []string{"lanes", "asks", "titles"} {
+		t.Run(lane, func(t *testing.T) {
+			rec := newAppRecorder()
+			pump, blocked := make(chan struct{}), make(chan struct{})
+			released := false
+			live := &appLive{
+				conn: &appConn{Release: func() { released = true }},
+				box:  appwire.NewOutbox(rec), pumpDone: pump,
+			}
+			switch lane {
+			case "lanes":
+				live.lanesDone = blocked
+			case "asks":
+				live.asksDone = blocked
+			case "titles":
+				live.titlesDone = blocked
+			}
+			// This also makes the pre-fix regression finish after its assertion.
+			defer close(blocked)
+			defer close(pump)
+			expired := make(chan struct{})
+			b := &appBackend{after: func(_ time.Duration, fn func()) func() {
+				timer := time.AfterFunc(80*time.Millisecond, func() {
+					fn()
+					close(expired)
+				})
+				return func() { timer.Stop() }
+			}}
+			live.box.Delta("agent", 1, 0, appwire.DeltaText, "last words")
+			done := make(chan struct{})
+			deadline := time.After(120 * time.Millisecond)
+			go func() { b.finish(live); close(done) }()
+			select {
+			case <-expired:
+				t.Log("drain timer callback completed with every producer still open")
+			case <-deadline:
+				t.Fatal("release drain timer did not fire through the shortened timer hook")
+			}
+			select {
+			case <-done:
+			case <-deadline:
+				t.Fatal("release waited beyond its one drain deadline")
+			}
+			if !released {
+				t.Fatal("conversation was not released")
+			}
+			rec.mu.Lock()
+			before := len(rec.events)
+			rec.mu.Unlock()
+			if before != 1 {
+				t.Fatalf("outbox lost its final delta: %d events", before)
+			}
+			live.box.Event("after-close", nil)
+			// Close waits for delivery; an already closed outbox ignores the event.
+			live.box.Close()
+			rec.mu.Lock()
+			defer rec.mu.Unlock()
+			if len(rec.events) != before {
+				t.Fatal("finish returned without closing its outbox")
+			}
+		})
+	}
+}
+
+// Contract 5: a clean release waits for every producer, flushes every event,
+// and returns promptly instead of waiting out the unused grace period.
+func TestAppFinishDrainsCleanProducersPromptly(t *testing.T) {
+	rec := newAppRecorder()
+	released := make(chan struct{})
+	live := &appLive{
+		conn:     &appConn{Release: func() { close(released) }},
+		box:      appwire.NewOutbox(rec),
+		pumpDone: make(chan struct{}), lanesDone: make(chan struct{}),
+		asksDone: make(chan struct{}), titlesDone: make(chan struct{}),
+	}
+	b := &appBackend{after: func(_ time.Duration, fn func()) func() {
+		timer := time.AfterFunc(time.Second, fn)
+		return func() { timer.Stop() }
+	}}
+	done := make(chan struct{})
+	go func() { b.finish(live); close(done) }()
+	<-released
+	for i, producer := range []chan struct{}{live.pumpDone, live.lanesDone, live.asksDone, live.titlesDone} {
+		// Give finish time to return incorrectly while this producer is open.
+		select {
+		case <-done:
+			t.Fatalf("finish returned before producer %d completed", i)
+		case <-time.After(25 * time.Millisecond):
+		}
+		live.box.Event(appwire.EventAgentChanged, appwire.AgentChanged{Agent: appwire.Agent{ID: "final"}})
+		close(producer)
+	}
+	select {
+	case <-done:
+	case <-time.After(150 * time.Millisecond):
+		t.Fatal("clean release waited out its grace period")
+	}
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	if len(rec.events) != 4 {
+		t.Fatalf("clean release lost events: %v", rec.events)
+	}
+}