A worker that needs its task token to finish starting cannot be dispatched

internal/commands/connect_worker_unix_test.go

@@ -29,7 +29,7 @@ func taskOf(p driver.Process) connector.TaskStatus {
 func runningTree(t *testing.T) (*driver.Worker, int) {
 	t.Helper()
 	pidFile := filepath.Join(t.TempDir(), "child")
-	worker, err := driver.StartWorker(context.Background(), nil, driver.Scope{WorkDir: t.TempDir()},
+	worker, err := driver.StartWorker(context.Background(), driver.SessionConfig{Scope: driver.Scope{WorkDir: t.TempDir()}},
 		driver.Command{Path: "/bin/sh", Args: []string{"-c", "sleep 300 & echo $! > " + pidFile + "; wait"}, Env: []string{"PATH=/bin:/usr/bin"}})
 	if err != nil {
 		t.Fatalf("start a worker: %v", err)

internal/connector/dispatcher.go

@@ -635,7 +635,14 @@ func (d *Dispatcher) start(ctx context.Context, record Record) error {
 		return nil
 	}
 	p := session.Process()
-	// The token goes only to this worker's own process group.
+	// The token goes only to this worker's own process group, and the socket
+	// was armed for it as soon as the process existed (SessionConfig.Started,
+	// in sessionConfig below). This is the backstop for a driver that
+	// announced nothing — one whose session runs somewhere the connector
+	// cannot signal, which has no group to allow either, so it arms on the
+	// zero group and the socket refuses every peer. AllowGroup takes the
+	// first group it is given and ignores the rest, so where the driver did
+	// announce, this changes nothing.
 	tokens.AllowGroup(p.PGID)
 	if err := d.ledger.MarkRunning(settleCtx, launch.AttemptID, recordedProcess(p, session.ID())); err != nil {
 		_ = session.Close()
@@ -735,6 +742,16 @@ func (d *Dispatcher) sessionConfig(ctx context.Context, launch Launch, record Re
 	return driver.SessionConfig{
 		Cwd: launch.WorkDir,
 		Env: driver.BuildEnv(driver.BaseEnv, d.opts.Lookup, nil),
+		// The socket is armed the moment the worker's process exists, which
+		// is inside NewSession and before whatever handshake the driver runs
+		// on top of it (driver invariant 7). Arming it after NewSession
+		// returned is a deadlock for any agent whose handshake does not
+		// finish until its MCP servers have connected: the server waits for a
+		// token the connector will not hand over until the handshake that is
+		// blocking it has returned, and both sides wait out their timeouts.
+		// AllowGroup does not block and takes only the first group it is
+		// given.
+		Started: func(p driver.Process) { tokens.AllowGroup(p.PGID) },
 		MCPServers: []driver.MCPServer{{
 			Name:    MCPServerName,
 			Command: d.opts.MCP.Command,

internal/connector/dispatcher_arming_test.go

@@ -0,0 +1,169 @@
+//go:build unix
+
+package connector
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/basecamp/basecamp-cli/internal/connector/driver"
+)
+
+// handshakeTokenWait is how long the worker in these tests waits for the task
+// token before it gives up. It stands in for the bridge's own 30-second dial
+// and the driver's two-minute handshake, which is what a deadlocked start
+// actually waits out: long enough that a loaded runner cannot fail this by
+// being slow, short enough that a regression costs the suite seconds rather
+// than half a minute.
+const handshakeTokenWait = 5 * time.Second
+
+// A worker whose start does not finish until its MCP server has the task
+// token used to wait for a token the connector would not hand over until that
+// start had finished.
+//
+// The socket accepts nothing until AllowGroup names the worker's process
+// group. The connector named it only once Driver.NewSession had returned, and
+// for the ACP driver the whole handshake — the adapter's own /mcp read-back,
+// which is a real prompt turn — runs inside NewSession. An agent that starts
+// its MCP servers there and will not answer until they have connected is
+// waiting on a socket that is waiting on it. Neither side moves until the
+// bridge's 30-second dial or the driver's two-minute handshake runs out, and
+// every dispatch through such an adapter fails. That is not hypothetical: a
+// fake that bound at session/new did exactly this while the Codex harness row
+// was being built.
+//
+// The fix is the ordering: the socket is armed when the worker's PROCESS
+// exists, which the driver says as soon as it has forked (driver invariant 7),
+// not when its session is ready.
+func TestAHandshakeThatWaitsForItsTaskTokenIsNotDeadlocked(t *testing.T) {
+	fake := newFakeDriver()
+	// The worker's group is this test's own, so the handshake below may take
+	// the token from the socket the way the worker's MCP server would.
+	fake.process = driver.Process{PID: os.Getpid(), PGID: syscall.Getpgrp(), StartedAt: time.Now()}
+	// The badly-behaved adapter: it starts its MCP server while it is opening
+	// its session, and it does not answer until that server has connected —
+	// which, for the connector's bridge, means until it has been handed its
+	// task token.
+	taken := make(chan string, 1)
+	fake.handshake = func(cfg driver.SessionConfig) error {
+		token, err := takeTaskToken(declaredSocket(cfg), handshakeTokenWait)
+		if err != nil {
+			taken <- ""
+			return fmt.Errorf("%w: the session's MCP server was never handed its task token: %w", driver.ErrSessionUnverified, err)
+		}
+		taken <- token
+		return nil
+	}
+	h := newDispatchHarness(t, fake, func(o *DispatcherOptions) { o.PrivateDir = tokenDir(t) })
+	// The "worker's group" is this test's own: confirming it gone would kill
+	// the test.
+	h.d.confirmGroupGone = func(driver.Process, time.Duration) error { return nil }
+	admitOn(t, h.ledger, 1, "recording:1")
+	h.run(t)
+	h.attemptsEnded(t, 1)
+
+	token := <-taken
+	require.NotEmpty(t, token, "the handshake was handed its task token while it was still running")
+	require.Len(t, fake.sessions, 1, "and the start it was blocking finished")
+	assert.NotEmpty(t, fake.sessions[0].promptList(), "so the worker was prompted")
+}
+
+// The same ordering, said as the rule rather than as its symptom: the
+// connector knows the worker's process group before the driver has opened a
+// session on it, and arms the socket then. A driver that announced nothing
+// until it returned would leave the socket unarmed here, which is what the
+// deadlock is made of.
+func TestTheTokenSocketIsArmedBeforeTheSessionIsOpen(t *testing.T) {
+	fake := newFakeDriver()
+	fake.process = driver.Process{PID: os.Getpid(), PGID: syscall.Getpgrp(), StartedAt: time.Now()}
+	watch := &announcingDriver{Driver: fake, seen: make(chan driver.Process, 4)}
+	armed := make(chan bool, 1)
+	fake.onStart = func(cfg driver.SessionConfig) {
+		assert.NotNil(t, cfg.Started, "the dispatcher asks to be told when the worker exists")
+	}
+	fake.handshake = func(cfg driver.SessionConfig) error {
+		// Inside NewSession, after the worker's process exists: by here the
+		// dispatcher has been told, and the socket takes a connection from
+		// the worker's group.
+		token, err := takeTaskToken(declaredSocket(cfg), handshakeTokenWait)
+		armed <- err == nil && token != ""
+		return nil
+	}
+	h := newDispatchHarness(t, fake, func(o *DispatcherOptions) {
+		o.PrivateDir = tokenDir(t)
+		o.Driver = watch
+	})
+	h.d.confirmGroupGone = func(driver.Process, time.Duration) error { return nil }
+	admitOn(t, h.ledger, 1, "recording:1")
+	h.run(t)
+	h.attemptsEnded(t, 1)
+
+	require.Len(t, watch.seen, 1, "the worker is announced once, as it starts")
+	assert.Equal(t, syscall.Getpgrp(), (<-watch.seen).PGID, "by the process group the token is served to")
+	assert.True(t, <-armed, "and the socket was serving that group before the session was open")
+}
+
+// announcingDriver records what the driver under test was asked to announce,
+// and passes the announcement on. It is the seam the dispatcher's own
+// ordering is read at: what it saw, and when.
+type announcingDriver struct {
+	driver.Driver
+	seen chan driver.Process
+}
+
+func (d *announcingDriver) NewSession(ctx context.Context, cfg driver.SessionConfig) (driver.Session, error) {
+	started := cfg.Started
+	cfg.Started = func(p driver.Process) {
+		d.seen <- p
+		if started != nil {
+			started(p)
+		}
+	}
+	return d.Driver.NewSession(ctx, cfg)
+}
+
+// declaredSocket is the token socket the session's MCP server declaration
+// names, which is the only place the worker learns it.
+func declaredSocket(cfg driver.SessionConfig) string {
+	if len(cfg.MCPServers) == 0 {
+		return ""
+	}
+	args := cfg.MCPServers[0].Args
+	return args[len(args)-1]
+}
+
+// takeTaskToken takes the task token from the connector's socket as the
+// bridge does: dial, read one line, close.
+func takeTaskToken(socket string, wait time.Duration) (string, error) {
+	if socket == "" {
+		return "", errors.New("the session declares no token socket")
+	}
+	deadline := time.Now().Add(wait)
+	dialer := net.Dialer{Timeout: wait}
+	conn, err := dialer.DialContext(context.Background(), "unix", socket)
+	if err != nil {
+		return "", fmt.Errorf("the connector's token socket: %w", err)
+	}
+	defer func() { _ = conn.Close() }()
+	_ = conn.SetDeadline(deadline)
+	line, err := bufio.NewReaderSize(conn, 256).ReadString('\n')
+	token := strings.TrimSpace(line)
+	if token == "" {
+		if err == nil {
+			err = errors.New("empty")
+		}
+		return "", fmt.Errorf("the connector handed over no token: %w", err)
+	}
+	return token, nil
+}

internal/connector/dispatcher_test.go

@@ -36,7 +36,11 @@ type fakeDriver struct {
 	sessions []*fakeSession
 	// turn answers each prompt; nil means end_turn at once.
 	turn func(s *fakeSession, n int, prompt string) (driver.PromptResult, error)
-	made chan *fakeSession
+	// handshake is what the driver does with the worker it has just started
+	// and announced, before NewSession returns: the ACP driver's handshake,
+	// where a real one runs. An error fails the start.
+	handshake func(cfg driver.SessionConfig) error
+	made      chan *fakeSession
 }
 
 func newFakeDriver() *fakeDriver { return &fakeDriver{made: make(chan *fakeSession, 16)} }
@@ -60,6 +64,18 @@ func (d *fakeDriver) NewSession(_ context.Context, cfg driver.SessionConfig) (dr
 	s := &fakeSession{d: d, cfg: cfg, done: make(chan struct{}), updates: make(chan driver.Update), canceled: make(chan struct{}, 1)}
 	d.sessions = append(d.sessions, s)
 	d.mu.Unlock()
+	// The worker's process exists from here: every driver that starts one
+	// says so before it opens a session on top of it (driver invariant 7),
+	// and a fake that did not would prove the dispatcher's ordering against a
+	// driver no driver is.
+	if cfg.Started != nil {
+		cfg.Started(s.Process())
+	}
+	if d.handshake != nil {
+		if err := d.handshake(cfg); err != nil {
+			return nil, err
+		}
+	}
 	d.made <- s
 	return s, nil
 }

internal/connector/driver/acp/acp.go

@@ -232,7 +232,7 @@ func (d *Driver) open(ctx context.Context, cfg driver.SessionConfig, loadID stri
 		more.Env = append(more.Env, driver.EnvOf(server.Env)...)
 	}
 	red := driver.NewRedactor(cfg.Redaction.With(more))
-	worker, err := driver.StartWorker(ctx, cfg.Launcher, cfg.Scope, driver.Command{
+	worker, err := driver.StartWorker(ctx, cfg, driver.Command{
 		Path: d.opts.Binary, Args: append([]string{}, d.opts.Args...), Env: env, Dir: cfg.Cwd,
 	})
 	if err != nil {