From d3aa35c7a1ce70156289193ba83eab95370c4103 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Wed, 28 Oct 2026 13:10:33 -0400 Subject: [PATCH] app: bound host joins and release drains Share one five-second budget across desktop host discovery and welcomes. Return a retryable unreachable receipt, close failed sockets and clear the deadline on successful joins without changing terminal dialing policy. Drain conversation producers against one closing deadline channel. Cover silent hosts, retry, long turns and delayed clean producers with fakes. Verify timeout consumption with the shortened timer hook, and document the desktop behavior. Co-Authored-By: Claude Opus 5.5 (1M context) --- cmd/codeaf/app.go | 104 +++- cmd/codeaf/appfinish_test.go | 115 +++++ cmd/codeaf/apphostjoin_test.go | 475 ++++++++++++++++++ .../unreleased/0-app-host-join-budget.md | 19 + .../manual/chat/running-from-the-terminal.md | 7 + 5 files changed, 709 insertions(+), 11 deletions(-) create mode 100644 cmd/codeaf/appfinish_test.go create mode 100644 cmd/codeaf/apphostjoin_test.go create mode 100644 docs/changes/unreleased/0-app-host-join-budget.md diff --git a/cmd/codeaf/app.go b/cmd/codeaf/app.go index 2a44d74ad7..69eaacf17f 100644 --- a/cmd/codeaf/app.go +++ b/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,28 +417,55 @@ 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) } @@ -441,6 +473,49 @@ func appJoinExisting(ask appOpenAsk, launch *remote.LaunchShape) (*appConn, erro 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: diff --git a/cmd/codeaf/appfinish_test.go b/cmd/codeaf/appfinish_test.go new file mode 100644 index 0000000000..6dae77a542 --- /dev/null +++ b/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) + } +} diff --git a/cmd/codeaf/apphostjoin_test.go b/cmd/codeaf/apphostjoin_test.go new file mode 100644 index 0000000000..ca6b1c4086 --- /dev/null +++ b/cmd/codeaf/apphostjoin_test.go @@ -0,0 +1,475 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "path/filepath" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/appwire" + "github.com/Agent-Field/codeaf/internal/config" + "github.com/Agent-Field/codeaf/internal/enginehost" + "github.com/Agent-Field/codeaf/internal/filelock" + "github.com/Agent-Field/codeaf/internal/remote" + "github.com/Agent-Field/codeaf/internal/session" +) + +func appJoinTestBudget(t *testing.T, wait time.Duration) { + t.Helper() + old := appHostJoinWait + appHostJoinWait = wait + t.Cleanup(func() { appHostJoinWait = old }) +} + +// appJoinTestHost uses the same short home and socket layout as answeringHost. +// Cleanup closes accepted sockets too, so a regression cannot strand a test. +func appJoinTestHost(t *testing.T, workspace string, answer func(net.Conn)) { + t.Helper() + if _, err := enginehost.Dir(workspace); err != nil { + t.Fatal(err) + } + path, err := enginehost.SocketPath(workspace) + if err != nil { + t.Fatal(err) + } + listener, err := net.Listen("unix", path) + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var sockets []net.Conn + var workers sync.WaitGroup + accepted := make(chan struct{}) + go func() { + defer close(accepted) + for { + conn, err := listener.Accept() + if err != nil { + return + } + mu.Lock() + sockets = append(sockets, conn) + mu.Unlock() + workers.Add(1) + go func() { + defer workers.Done() + defer conn.Close() + answer(conn) + }() + } + }() + t.Cleanup(func() { + _ = listener.Close() + <-accepted + mu.Lock() + for _, conn := range sockets { + _ = conn.Close() + } + mu.Unlock() + workers.Wait() + }) +} + +func appJoinTestChat(t *testing.T) (workspace, transcript string) { + t.Helper() + shortStateRoot(t) + t.Setenv("HOME", t.TempDir()) + t.Setenv(config.ProfileDirEnv, t.TempDir()) + t.Setenv(config.APIKeyEnv, "") + t.Setenv("CODEAF_SYNC_URL", "off") + workspace = t.TempDir() + bucket, err := v3ProjectDir(workspace) + if err != nil { + t.Fatal(err) + } + dir := writeV3Session(t, bucket, "0123456789abcdef", "saved words", time.Now()) + if err := session.SaveMeta(dir, session.Meta{ID: "0123456789abcdef", Workspace: workspace, Created: time.Now()}); err != nil { + t.Fatal(err) + } + transcript = session.Place{Dir: dir}.Transcript() + file, err := os.OpenFile(transcript, os.O_RDWR, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = file.Close() }) + if err := filelock.Lock(file, true, true); err != nil { + t.Fatal(err) + } + return workspace, transcript +} + +// Contract 1: first open is bounded, independent reads answer, and the next +// open reaches the host again after the failed attempt releases its gate. +func TestAppHostJoinSilentOpenReleasesGate(t *testing.T) { + workspace, transcript := appJoinTestChat(t) + before, err := os.ReadFile(transcript) + if err != nil { + t.Fatal(err) + } + appJoinTestBudget(t, 80*time.Millisecond) + b := newFakeBackend(newAppRecorder(), appOpenInProcess) + b.profileDir = t.TempDir() + t.Cleanup(func() { _ = b.Close() }) + hellos := make(chan remote.Hello, 4) + closed := make(chan int64, 4) + appJoinTestHost(t, workspace, func(conn net.Conn) { + var frame remote.Frame + if err := json.NewDecoder(conn).Decode(&frame); err != nil { + return + } + var hello remote.Hello + _ = json.Unmarshal(frame.Payload, &hello) + hellos <- hello + n, _ := io.Copy(io.Discard, conn) + closed <- n + }) + for attempt := range 4 { + started := time.Now() + opened := make(chan error, 1) + go func() { _, err := b.OpenAgent(appAgentID(transcript)); opened <- err }() + select { + case hello := <-hellos: + if !hello.Join || hello.Session != transcript || hello.New { + t.Fatalf("open was not a join: %+v", hello) + } + case <-time.After(time.Second): + t.Fatal("open did not reach the host") + } + reads := make(chan error, 1) + go func() { + _, err := b.ListAgents("") + if err == nil { + _, err = b.ListModels() + } + reads <- err + }() + select { + case err := <-reads: + if err != nil { + t.Fatal(err) + } + case <-time.After(150 * time.Millisecond): + t.Fatal("agent/model reads waited behind the silent host") + } + select { + case err := <-opened: + t.Fatalf("open finished before the concurrent reads: %v", err) + default: + } + select { + case err := <-opened: + var wire *appwire.Error + if !errors.As(err, &wire) || wire.Code != appwire.CodeUnreachable || wire.Message != "This chat didn’t answer from the other codeaf window. Try again, or continue there." { + t.Fatalf("attempt %d: open error = %v", attempt, err) + } + case <-time.After(time.Until(started.Add(200 * time.Millisecond))): + t.Fatal("first open exceeded the shared join budget plus slack") + } + t.Logf("silent open %d: %s", attempt+1, time.Since(started)) + select { + case n := <-closed: + if n != 0 { + t.Fatalf("failed open sent %d bytes after hello", n) + } + case <-time.After(time.Second): + t.Fatal("failed open left its socket open") + } + if !session.InUse(transcript) { + t.Fatal("failed open took the journal from its original writer") + } + } + if files, err := filepath.Glob(filepath.Join(session.DirOf(transcript), "tasks", "*")); err != nil || len(files) != 0 { + t.Fatalf("failed open created tasks: %v, %v", files, err) + } + if after, err := os.ReadFile(transcript); err != nil || !bytes.Equal(before, after) { + t.Fatalf("failed opens changed the journal: %v", err) + } +} + +// Contract 2: discovery has one deadline even after earlier candidates have +// used part of it, and exhaustion never moves on to another silent socket. +func TestAppHostJoinCandidatesShareBudget(t *testing.T) { + for _, refusals := range []int32{0, 2} { + t.Run(fmt.Sprintf("refusals=%d", refusals), func(t *testing.T) { + workspace, transcript := appJoinTestChat(t) + appJoinTestBudget(t, 120*time.Millisecond) + var attempts atomic.Int32 + closed := make(chan struct{}, 4) + for range 4 { + appJoinTestHost(t, t.TempDir(), func(conn net.Conn) { + var hello remote.Frame + if json.NewDecoder(conn).Decode(&hello) != nil { + return + } + if attempts.Add(1) <= refusals { + time.Sleep(35 * time.Millisecond) + _ = json.NewEncoder(conn).Encode(remote.Frame{Kind: "fatal", Error: "not held here"}) + } + _, _ = io.Copy(io.Discard, conn) + closed <- struct{}{} + }) + } + started := time.Now() + opened := make(chan error, 1) + go func() { + _, err := appJoinExisting(appOpenAsk{Workspace: workspace, Transcript: transcript}, nil) + opened <- err + }() + select { + case err := <-opened: + var wire *appwire.Error + if !errors.As(err, &wire) || wire.Code != appwire.CodeUnreachable { + t.Fatalf("join error = %v", err) + } + case <-time.After(170 * time.Millisecond): + t.Fatal("candidate handshakes multiplied the overall join budget") + } + t.Logf("shared join: %s, %d candidates", time.Since(started), attempts.Load()) + if got := attempts.Load(); got != refusals+1 { + t.Fatalf("asked %d candidates, want %d", got, refusals+1) + } + for range refusals + 1 { + select { + case <-closed: + case <-time.After(time.Second): + t.Fatal("a candidate socket was not closed") + } + } + }) + } +} + +type appJoinScriptedAgent struct{ scriptedAgent } + +func (a *appJoinScriptedAgent) ModelIdentity() session.ModelIdentity { + return session.ModelIdentity{Selected: "a/b", Models: &session.ConversationModels{Main: "a/b"}} +} + +// Contracts 3 and 4: Try again joins the same live session without a send or +// takeover. A subsequent explicit turn survives beyond the handshake budget. +func TestAppHostJoinRetryAndLongTurn(t *testing.T) { + workspace, transcript := appJoinTestChat(t) + appJoinTestBudget(t, 40*time.Millisecond) + agent := &appJoinScriptedAgent{scriptedAgent: scriptedAgent{file: transcript}} + sess := remote.NewSession(&remote.Engine{Agent: agent, Workspace: workspace, SessionFile: transcript}, true) + t.Cleanup(func() { _ = sess.Close() }) + var ready atomic.Bool + closed := make(chan struct{}, 1) + b := newFakeBackend(newAppRecorder(), appOpenInProcess) + b.profileDir = t.TempDir() + t.Cleanup(func() { _ = b.Close() }) + appJoinTestHost(t, workspace, func(conn net.Conn) { + if !ready.Load() { + _, _ = io.Copy(io.Discard, conn) + closed <- struct{}{} + return + } + _ = remote.ServeAttach(conn, conn, remote.AttachOptions{ + Open: func(hello remote.Hello) (*remote.Session, error) { + if !hello.Join || hello.New || hello.Session != transcript { + return nil, fmt.Errorf("expected existing conversation join: %+v", hello) + } + return sess, nil + }, + }) + }) + if _, err := b.OpenAgent(appAgentID(transcript)); err == nil { + t.Fatal("silent host unexpectedly opened") + } + select { + case <-closed: + case <-time.After(time.Second): + t.Fatal("failed open left a socket behind") + } + ready.Store(true) + view, err := b.OpenAgent(appAgentID(transcript)) + if err != nil || view.ID != appAgentID(transcript) { + t.Fatalf("Try again failed to open the same chat: %+v, %v", view, err) + } + agent.mu.Lock() + sends, interrupts, closes := len(agent.said), agent.interrupts, agent.closes + agent.mu.Unlock() + if sends != 0 || interrupts != 0 || closes != 0 || !session.InUse(transcript) { + t.Fatalf("open mutated the host: sends=%d interrupts=%d closes=%d", sends, interrupts, closes) + } + a, err := b.find(view.ID) + if err != nil { + t.Fatal(err) + } + live := b.liveOf(a) + if live.conn.Transcript != transcript || live.conn.Assembled { + t.Fatal("retry did not join the original host conversation") + } + events, err := live.conn.Conv.Submit(context.Background(), "one deliberate turn") + if err != nil { + t.Fatal(err) + } + select { + case <-events: + case <-time.After(time.Second): + t.Fatal("joined turn never started") + } + time.Sleep(2 * appHostJoinWait) + agent.mu.Lock() + turn := agent.turn + turn <- session.Event{Kind: session.EventTextDelta, Text: "past the join budget"} + close(turn) + agent.turn = nil + agent.mu.Unlock() + seen, ended := false, false + deadline := time.After(time.Second) + for !ended { + select { + case event, ok := <-events: + ended = !ok + seen = seen || event.Text == "past the join budget" + case <-deadline: + t.Fatal("long turn did not finish after its producer closed") + } + } + if !seen { + t.Fatal("handshake deadline interrupted the long turn") + } + if _, err := b.OpenAgent(view.ID); err != nil { + t.Fatalf("joined conversation failed after the budget: %v", err) + } +} + +// Contract 4: refusals, incompatible welcomes and truncated/closed welcomes +// return promptly without accumulating sockets or handshake goroutines. +func TestAppHostJoinFailedCandidatesClose(t *testing.T) { + for _, mode := range []string{"refused", "incompatible", "partial", "eof"} { + t.Run(mode, func(t *testing.T) { + workspace, transcript := appJoinTestChat(t) + appJoinTestBudget(t, 100*time.Millisecond) + closed := make(chan struct{}, 32) + appJoinTestHost(t, workspace, func(conn net.Conn) { + var hello remote.Frame + if json.NewDecoder(conn).Decode(&hello) != nil { + return + } + switch mode { + case "refused": + _ = json.NewEncoder(conn).Encode(remote.Frame{Kind: "fatal", Error: "not held here"}) + case "incompatible": + payload, _ := json.Marshal(remote.Welcome{Version: remote.Version + 1}) + _ = json.NewEncoder(conn).Encode(remote.Frame{Kind: "welcome", Payload: payload}) + case "partial": + _, _ = io.WriteString(conn, `{"kind":"welcome","payload":`) + _ = conn.Close() + case "eof": + _ = conn.Close() + } + _, _ = io.Copy(io.Discard, conn) + closed <- struct{}{} + }) + open := func() { + t.Helper() + started := time.Now() + conn, err := appJoinExisting(appOpenAsk{Workspace: workspace, Transcript: transcript}, nil) + if conn != nil { + conn.Release() + t.Fatal("a refused host was joined") + } + var wire *appwire.Error + if !errors.As(err, &wire) || wire.Code != appwire.CodeConflict { + t.Fatalf("a prompt refusal lost its existing conflict: %v", err) + } + if elapsed := time.Since(started); elapsed >= appHostJoinWait { + t.Fatalf("a failed welcome spent the whole budget: %s", elapsed) + } + for range 2 { + select { + case <-closed: + case <-time.After(time.Second): + t.Fatal("a failed welcome left its socket open") + } + } + } + open() + fds, fdErr := os.ReadDir("/proc/self/fd") + goroutines := runtime.NumGoroutine() + for range 8 { + open() + } + until := time.Now().Add(100 * time.Millisecond) + for { + after, err := os.ReadDir("/proc/self/fd") + fdLeak := fdErr == nil && err == nil && len(after) > len(fds)+1 + goroutineLeak := runtime.NumGoroutine() > goroutines+2 + if !fdLeak && !goroutineLeak { + break + } + if time.Now().After(until) { + t.Fatalf("failed joins retained resources: fds %d -> %d; goroutines %d -> %d", len(fds), len(after), goroutines, runtime.NumGoroutine()) + } + time.Sleep(time.Millisecond) + } + }) + } +} + +// Contracts 1 and 2: a starting host's remove-to-listen grace also shares the +// join deadline. An unheld stale socket still answers without that grace. +func TestAppHostJoinStartingHostSharesBudget(t *testing.T) { + for _, held := range []bool{false, true} { + t.Run(fmt.Sprintf("host-lock=%t", held), func(t *testing.T) { + workspace, transcript := appJoinTestChat(t) + appJoinTestBudget(t, 60*time.Millisecond) + if _, err := enginehost.Dir(workspace); err != nil { + t.Fatal(err) + } + path, err := enginehost.SocketPath(workspace) + if err != nil { + t.Fatal(err) + } + stale, err := net.Listen("unix", path) + if err != nil { + t.Fatal(err) + } + stale.(*net.UnixListener).SetUnlinkOnClose(false) + _ = stale.Close() + lockPath, err := enginehost.LockPath(workspace) + if err != nil { + t.Fatal(err) + } + file, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatal(err) + } + defer file.Close() + if held { + if err := filelock.Lock(file, true, true); err != nil { + t.Fatal(err) + } + } + started := time.Now() + _, err = appJoinExisting(appOpenAsk{Workspace: workspace, Transcript: transcript}, nil) + var wire *appwire.Error + want := appwire.CodeConflict + if held { + want = appwire.CodeUnreachable + } + if !errors.As(err, &wire) || wire.Code != want { + t.Fatalf("starting host error = %v, want %s", err, want) + } + limit := 100 * time.Millisecond + if !held { + limit = 40 * time.Millisecond + } + if elapsed := time.Since(started); elapsed > limit { + t.Fatalf("starting/stale host exceeded its budget: %s", elapsed) + } + }) + } +} diff --git a/docs/changes/unreleased/0-app-host-join-budget.md b/docs/changes/unreleased/0-app-host-join-budget.md new file mode 100644 index 0000000000..6b9166bdee --- /dev/null +++ b/docs/changes/unreleased/0-app-host-join-budget.md @@ -0,0 +1,19 @@ +--- +kind: fixed +title: Bound desktop chat joins and conversation release drains +pr: 0 +surface: [engine, docs] +invalidates: + - "A desktop chat open could wait indefinitely for a terminal host's welcome. All candidate hosts now share one five-second join budget and a timeout offers Try again." + - "A conversation release could consume its timeout while waiting for the turn pump and then wait forever for another producer. All event producers now share one release deadline." +--- + +An unresponsive host answers the desktop with `unreachable` and “This chat +didn’t answer from the other codeaf window. Try again, or continue there.” +Failed opens close their sockets without taking over the journal or sending a +message. Successful joins clear their handshake deadline so long replies keep +running. The terminal's own handshake behavior is unchanged. + +A clean conversation release still drains every producer and flushes queued +events promptly. If the shared release deadline expires, later unfinished +producers cannot hold the outbox open indefinitely. diff --git a/internal/manual/chat/running-from-the-terminal.md b/internal/manual/chat/running-from-the-terminal.md index 897e0b7586..47fda3a0da 100644 --- a/internal/manual/chat/running-from-the-terminal.md +++ b/internal/manual/chat/running-from-the-terminal.md @@ -614,6 +614,13 @@ conversation lives in another folder's terminal (a crew member a manager started). If it cannot reach that terminal it says the conversation is open in another codeaf window rather than opening it twice. +If that window accepts a connection but does not answer, the app waits up to +five seconds across all candidate windows, then shows “Couldn’t open chat”: +“This chat didn’t answer from the other codeaf window. Try again, or continue +there.” Choose “Try again” when that window answers again. A failed open sends +no message, takes over no work and leaves the terminal's conversation alone. +The five-second limit ends with the handshake; joined replies can run longer. + A joined conversation keeps the rules of the window that opened it, because they are how it behaves for everyone: a proposed task's clock, a choice's clock, moving a running reply to a task, naming, and saving proposed reminders