From 21df8f82c4656408b68354051fc753f4bb61fc8a Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:36:13 +0200 Subject: [PATCH] Arm the task token's socket when the worker exists, not when its handshake returns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The socket accepts nothing until AllowGroup names the worker's process group, and the connector named it only once Driver.NewSession had returned. For the ACP driver the whole handshake runs inside NewSession, the adapter's own /mcp read-back — a real prompt turn — included. An adapter that will not finish that handshake until its MCP servers have connected therefore waits for a token the connector cannot hand over until the handshake blocking it has finished, and both sides sit there until the bridge's 30-second dial or the driver's two-minute handshake runs out. A fake that bound at session/new hit exactly this while the Codex harness row was being built. SessionConfig.Started announces the worker's process the moment the fork returns. It is called from driver.StartWorker, the one place every driver starts its agent, so no driver can forget it; the dispatcher wires it to AllowGroup. Written up as driver invariant 7. The dispatcher still arms on Session.Process() afterwards as a backstop for a driver that announces nothing — AllowGroup takes only the first group it is given, so where the driver did announce, that changes nothing. What the earlier arming widens is the token window, and that is the one thing that relied on the old order: it starts at AllowGroup, so the handshake is inside it now. The card-23 review's concern was a launcher that takes its time, which is preserved — the arming is after the fork. The window is no smaller for the case it is for, since an agent starts its MCP servers during its handshake and their connection is already waiting in the listener's backlog. And it is deliberately not stretched to cover both: the socket is armed for the worker's whole process group, which is the agent's own tree, so a longer armed window is a longer window in which the agent's tools could ask for the token instead of its MCP server. DefaultTokenWindow now says that rather than the old order. The compat check armed on Session.Process() right after NewSession under either ordering, so it passed under both and would not have caught a regression. It arms from cfg.Started now, where the dispatcher does, and fails outright on the old order. Since it cannot run without the pinned adapters and model quota, its verdict is a small object driven through both orderings against the real driver in the ordinary suite (TestTheArmingCheckTellsTheTwoOrderingsApart). --- internal/commands/connect_worker_unix_test.go | 2 +- internal/connector/dispatcher.go | 19 +- internal/connector/dispatcher_arming_test.go | 169 ++++++++++++++++++ internal/connector/dispatcher_test.go | 18 +- internal/connector/driver/acp/acp.go | 2 +- internal/connector/driver/acp/arming_test.go | 165 +++++++++++++++++ internal/connector/driver/acp/compat_test.go | 47 ++++- internal/connector/driver/announce_test.go | 50 ++++++ internal/connector/driver/claude/claude.go | 2 +- internal/connector/driver/codex/codex.go | 2 +- internal/connector/driver/driver.go | 24 ++- internal/connector/driver/driver_test.go | 18 +- .../connector/driver/drivertest/drivertest.go | 2 +- internal/connector/driver/worker.go | 25 ++- internal/connector/driver/worker_other.go | 2 +- internal/connector/recovery_acp_test.go | 28 +-- internal/connector/tokensocket.go | 19 +- 17 files changed, 551 insertions(+), 43 deletions(-) create mode 100644 internal/connector/dispatcher_arming_test.go create mode 100644 internal/connector/driver/acp/arming_test.go create mode 100644 internal/connector/driver/announce_test.go diff --git a/internal/commands/connect_worker_unix_test.go b/internal/commands/connect_worker_unix_test.go index 2f8d780f1..502ba22f4 100644 --- a/internal/commands/connect_worker_unix_test.go +++ b/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) diff --git a/internal/connector/dispatcher.go b/internal/connector/dispatcher.go index 440514a8e..07332fdb9 100644 --- a/internal/connector/dispatcher.go +++ b/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, diff --git a/internal/connector/dispatcher_arming_test.go b/internal/connector/dispatcher_arming_test.go new file mode 100644 index 000000000..e0e27ee74 --- /dev/null +++ b/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 +} diff --git a/internal/connector/dispatcher_test.go b/internal/connector/dispatcher_test.go index 6111e6d54..e2df31999 100644 --- a/internal/connector/dispatcher_test.go +++ b/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 } diff --git a/internal/connector/driver/acp/acp.go b/internal/connector/driver/acp/acp.go index b1c9d8690..7340b8e9e 100644 --- a/internal/connector/driver/acp/acp.go +++ b/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 { diff --git a/internal/connector/driver/acp/arming_test.go b/internal/connector/driver/acp/arming_test.go new file mode 100644 index 000000000..0b1b07b01 --- /dev/null +++ b/internal/connector/driver/acp/arming_test.go @@ -0,0 +1,165 @@ +//go:build unix + +package acp + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" +) + +// tokenArming is when the connector was told the worker's process group, +// relative to the start it was told it during. It is what the adapter +// compatibility check reads (compat_test.go, check 7): that check used to arm +// the token socket on Session.Process() right after NewSession returned, which +// is what it does under either ordering, so it passed whichever one the +// connector used and could not have caught a regression in it. +// +// TestTheArmingCheckTellsTheTwoOrderingsApart drives this through both +// orderings against the real driver, so the compat check's verdict is one that +// has been shown to distinguish them rather than one assumed to. +type tokenArming struct { + mu sync.Mutex + armed bool + group int + duringStart bool + returned bool +} + +// arm is what SessionConfig.Started is given: the worker's process, as soon +// as it exists. +func (a *tokenArming) arm(p driver.Process) { + a.mu.Lock() + defer a.mu.Unlock() + if a.armed { + return + } + a.armed, a.group, a.duringStart = true, p.PGID, !a.returned +} + +// startReturned is called the moment NewSession returns. +func (a *tokenArming) startReturned() { + a.mu.Lock() + defer a.mu.Unlock() + a.returned = true +} + +// verdict says whether the socket was armed in time for an agent whose +// handshake does not finish until its MCP servers have connected. +func (a *tokenArming) verdict() error { + a.mu.Lock() + defer a.mu.Unlock() + switch { + case !a.armed: + return errors.New("the connector was never told the worker's process group, so the token socket served nobody") + case !a.duringStart: + return errors.New("the token socket was armed only after NewSession returned: an agent that cannot finish its handshake until its MCP servers have connected waits for a token this ordering cannot hand over until that handshake has finished") + case a.group <= 1: + return fmt.Errorf("the worker was announced as process group %d, which the token socket serves nothing", a.group) + } + return nil +} + +// The compat check's verdict must come out differently under the two +// orderings, or it is measuring nothing about either. Both runs use the real +// driver and the same fake adapter; only where the arming happens differs. +func TestTheArmingCheckTellsTheTwoOrderingsApart(t *testing.T) { + t.Run("armed as soon as the worker's process exists", func(t *testing.T) { + h := newHarness(t) + var arming tokenArming + h.withConfig = func(cfg driver.SessionConfig) driver.SessionConfig { + cfg.Started = arming.arm + return cfg + } + s, err := h.driver().NewSession(context.Background(), h.config()) + arming.startReturned() + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + assert.NoError(t, arming.verdict(), "the ordering the connector uses now") + }) + + t.Run("armed once the session was open", func(t *testing.T) { + h := newHarness(t) + var arming tokenArming + s, err := h.driver().NewSession(context.Background(), h.config()) + arming.startReturned() + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + // The ordering the connector used to use, and the one the compat + // check used to perform on its own: Session.Process(), after the + // handshake. + arming.arm(s.Process()) + assert.ErrorContains(t, arming.verdict(), "after NewSession returned", + "the check must fail on the ordering the card is about") + }) +} + +// This driver is the one the handshake ordering is about: initialize, the +// session, its asking mode and the adapter's own /mcp read-back — a real +// prompt turn — all run inside NewSession. An adapter that will not finish +// that handshake until its MCP servers have connected is waiting for a task +// token, and the connector cannot hand one over until it knows the worker's +// process group. So the group must be known while the handshake is still +// running, not when it returns (driver invariant 7). +// +// The fake hangs at session/new, which is the middle of the handshake and +// stands in for an adapter waiting on its servers. The worker must already +// have been announced by then. +func TestTheWorkerIsAnnouncedWhileTheHandshakeIsStillRunning(t *testing.T) { + h := newHarness(t) + h.sc.Hang = "session/new" + announced := make(chan driver.Process, 1) + h.withConfig = func(cfg driver.SessionConfig) driver.SessionConfig { + cfg.Started = func(p driver.Process) { announced <- p } + return cfg + } + d := h.driver() + d.opts.HandshakeTimeout = 3 * time.Second + + type start struct { + s driver.Session + err error + } + done := make(chan start, 1) + go func() { + s, err := d.NewSession(context.Background(), h.config()) + done <- start{s, err} + }() + + var worker driver.Process + select { + case worker = <-announced: + case got := <-done: + if got.s != nil { + _ = got.s.Close() + } + t.Fatalf("the handshake ended without the worker ever being announced: %v", got.err) + case <-time.After(30 * time.Second): + t.Fatal("the worker was never announced") + } + assert.Greater(t, worker.PGID, 1, "announced by a process group the token socket can serve") + assert.Equal(t, worker.PID, worker.PGID, "the adapter leads its own group") + select { + case got := <-done: + if got.s != nil { + _ = got.s.Close() + } + t.Fatalf("the handshake had already finished: this proves nothing about the order (%v)", got.err) + default: + } + + // And the hung handshake ends as it always did: a start that launched a + // process, whose group the driver has already asked to end. + got := <-done + require.Error(t, got.err) + require.Nil(t, got.s) + assert.Equal(t, worker, driver.StartedProcess(got.err), "the process the failed start names is the one it announced") +} diff --git a/internal/connector/driver/acp/compat_test.go b/internal/connector/driver/acp/compat_test.go index 69cebef81..7632bb9e2 100644 --- a/internal/connector/driver/acp/compat_test.go +++ b/internal/connector/driver/acp/compat_test.go @@ -565,8 +565,10 @@ func checkDecoyMCPServer(t *testing.T, e compatEnv) { // Check 7: the task token's carriage, as the dispatcher builds it. The MCP // server is the connector's bridge (`basecamp connect worker-mcp`), the token // is served once on a socket in the attempt's private directory, and the -// socket is told the worker's process group only once NewSession returns — -// the order the dispatcher uses. The bridge must reach the socket from +// socket is told the worker's process group as soon as that process exists, +// while the adapter is still in its handshake — the order the dispatcher +// uses, and the order this check now holds it to rather than performing on +// its own after the fact. The bridge must reach the socket from // wherever the adapter starts it, the handoff must be delivered, and the // token must not be in any environment, command line or file of the worker's // processes. No Basecamp account is involved: the bridge's profile is a dummy @@ -621,19 +623,54 @@ func checkTokenBridge(t *testing.T, e compatEnv) { Scope: driver.Scope{WorkDir: wd}, PrivateDir: private, } + // The arming is the dispatcher's, done where the dispatcher does it: on + // the worker's process, as soon as that process exists and while the + // adapter is still in its handshake. Arming it after NewSession returned + // — which is what this check used to do — passes under either ordering + // and so says nothing about either; arming it here fails outright on the + // ordering the card is about, because an adapter that started its MCP + // servers during its handshake would then be waiting for a token nothing + // had armed the socket to hand over. tokenArming.verdict is the + // difference, and TestTheArmingCheckTellsTheTwoOrderingsApart proves it + // is one. + var arming tokenArming + cfg.Started = func(p driver.Process) { + arming.arm(p) + tokens.AllowGroup(p.PGID) + } + // Read before the start, not after it: the handoff this check is about + // can now happen while NewSession is still running, and a result read + // afterwards could not say whether it did. + handed := make(chan connector.Handoff, 1) + go func() { handed <- tokens.Result() }() + d := e.driverFor(t, "") var s driver.Session var places drivertest.Places drivertest.RequireNoSecretFilesDuring(t, token, []string{wd, private, state}, func() { started := time.Now() s, err = d.NewSession(turnCtx(t), cfg) + arming.startReturned() if err != nil { t.Fatalf("NewSession: %v", err) } t.Logf("NewSession took %s", time.Since(started).Round(time.Millisecond)) - tokens.AllowGroup(s.Process().PGID) - handed := make(chan connector.Handoff, 1) - go func() { handed <- tokens.Result() }() + if err := arming.verdict(); err != nil { + _ = s.Close() + t.Fatalf("the task token's socket was not armed in time: %v", err) + } + delivered := "" + select { + case h := <-handed: + // Taken while the adapter was still opening its session: this is + // the case the ordering exists for, and the old one could not + // have reached it. + handed <- h + delivered = "during the handshake" + default: + delivered = "after the handshake" + } + t.Logf("the token was taken %s", delivered) deadline := time.After(90 * time.Second) for { places = addWorkerProcesses(places, s.Process().PID) diff --git a/internal/connector/driver/announce_test.go b/internal/connector/driver/announce_test.go new file mode 100644 index 000000000..f6d2c9c2a --- /dev/null +++ b/internal/connector/driver/announce_test.go @@ -0,0 +1,50 @@ +//go:build unix + +package driver + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Invariant 7: the worker is announced as soon as it exists, not when a +// session is ready on top of it. StartWorker is where every driver's worker +// is started, so it is where the announcement is made — a driver cannot +// forget it, and the connector cannot be left arming a token socket for a +// process group it will not be told about until a handshake that is waiting +// on that socket has returned. +func TestAWorkerIsAnnouncedAsSoonAsItsProcessExists(t *testing.T) { + var announced []Process + cfg := SessionConfig{ + Scope: Scope{WorkDir: t.TempDir()}, + Started: func(p Process) { announced = append(announced, p) }, + } + w, err := StartWorker(context.Background(), cfg, Command{Path: "/bin/sleep", Args: []string{"30"}, Env: []string{}}) + require.NoError(t, err) + t.Cleanup(func() { w.Terminate(time.Second) }) + + require.Len(t, announced, 1, "announced once, by the time StartWorker returns") + assert.Equal(t, w.Process(), announced[0], "and it is the worker StartWorker is handing back") + assert.Positive(t, announced[0].PID) + assert.Equal(t, announced[0].PID, announced[0].PGID, "which leads its own group, so the token socket can be told one number") +} + +// A start that never made a process has nothing to announce: the connector +// would otherwise arm a socket for a group that does not exist. +func TestAStartThatLaunchedNothingAnnouncesNothing(t *testing.T) { + announced := 0 + cfg := SessionConfig{ + Scope: Scope{WorkDir: t.TempDir()}, + Started: func(Process) { announced++ }, + } + _, err := StartWorker(context.Background(), cfg, Command{Path: "/nonexistent/agent-not-here"}) + require.ErrorIs(t, err, ErrNotStarted) + _, err = StartWorker(context.Background(), SessionConfig{Launcher: refusingLauncher{}, Scope: cfg.Scope, Started: cfg.Started}, + Command{Path: "/bin/true"}) + require.ErrorIs(t, err, ErrNotStarted) + assert.Zero(t, announced) +} diff --git a/internal/connector/driver/claude/claude.go b/internal/connector/driver/claude/claude.go index 002dc9877..d5264081d 100644 --- a/internal/connector/driver/claude/claude.go +++ b/internal/connector/driver/claude/claude.go @@ -200,7 +200,7 @@ func (d *Driver) start(ctx context.Context, cfg driver.SessionConfig, sessionID return nil, fmt.Errorf("%w: %w: %w", driver.ErrNotStarted, driver.ErrUnusable, err) } env := d.env(cfg) - worker, err := driver.StartWorker(ctx, cfg.Launcher, cfg.Scope, driver.Command{Path: d.opts.Binary, Args: args, Env: env, Dir: cfg.Cwd}) + worker, err := driver.StartWorker(ctx, cfg, driver.Command{Path: d.opts.Binary, Args: args, Env: env, Dir: cfg.Cwd}) if err != nil { _ = os.Remove(mcpPath) return nil, err diff --git a/internal/connector/driver/codex/codex.go b/internal/connector/driver/codex/codex.go index 0b6901504..083838ae1 100644 --- a/internal/connector/driver/codex/codex.go +++ b/internal/connector/driver/codex/codex.go @@ -313,7 +313,7 @@ func (d *Driver) start(ctx context.Context, cfg driver.SessionConfig, resumeID s if err != nil { return nil, fmt.Errorf("%w: %w", driver.ErrNotStarted, err) } - worker, err := driver.StartWorker(ctx, cfg.Launcher, cfg.Scope, driver.Command{Path: d.opts.Binary, Args: args, Env: env, Dir: cfg.Cwd}) + worker, err := driver.StartWorker(ctx, cfg, driver.Command{Path: d.opts.Binary, Args: args, Env: env, Dir: cfg.Cwd}) if err != nil { return nil, err } diff --git a/internal/connector/driver/driver.go b/internal/connector/driver/driver.go index 13fd9fbda..0b9aa7681 100644 --- a/internal/connector/driver/driver.go +++ b/internal/connector/driver/driver.go @@ -49,6 +49,15 @@ // counts; they never carry the agent's text or a tool's input, so a sink // that logs an update cannot log content. What a sink does log from an // agent stream goes through the redaction rule (redact.go). +// 7. A worker is announced as soon as it exists, not when its session is +// ready. SessionConfig.Started is called with the worker's process the +// moment the fork returns, before any handshake the driver runs — which +// is what lets the connector arm the task token's socket for a worker +// whose handshake cannot finish until its MCP servers have connected. +// StartWorker calls it, so every driver that starts a process holds this +// by construction; a driver whose session runs somewhere the connector +// cannot signal announces nothing, and the connector arms on what +// Session.Process reports instead. // // # Refusals: where one is recorded, and when it counts as settled // @@ -115,7 +124,9 @@ type Driver interface { Name() string // Capabilities says what the driver supports beyond NewSession and Prompt. Capabilities() Capabilities - // NewSession starts a worker and opens a session in cfg.Cwd. + // NewSession starts a worker and opens a session in cfg.Cwd. The worker + // is announced through cfg.Started as soon as its process exists, before + // the session is opened on top of it (invariant 7). // // An error that wraps ErrNotStarted means no process ever existed, and // the connector may retry the start once. Any other error from a start @@ -207,6 +218,17 @@ type SessionConfig struct { // Refusals records every refusal at the moment it is made or observed. // Nil records nothing; the dispatcher always sets it. Refusals RefusalRecorder + // Started is called once with the worker's process as soon as that + // process exists — the fork has returned and the group is the worker's — + // and before whatever handshake the driver runs on top of it (invariant + // 7). It is how the connector learns the process group while NewSession + // is still running, which is what arms the task token's socket in time + // for an agent that starts its MCP servers during its own handshake: + // arming only once NewSession returned deadlocks such an agent against a + // token the connector cannot hand over until the handshake it is blocking + // has finished. It must not block: a driver calls it on the goroutine + // that is starting the session. Nil announces nothing. + Started func(Process) // Redaction is what the driver takes out of every error it returns and // every text an update or a stderr tail carries (redact.go). The driver // adds the environment it builds, its MCP servers' environments and diff --git a/internal/connector/driver/driver_test.go b/internal/connector/driver/driver_test.go index af1c70bc4..71740bede 100644 --- a/internal/connector/driver/driver_test.go +++ b/internal/connector/driver/driver_test.go @@ -34,7 +34,7 @@ func TestBuildEnvTakesExactNamesOnly(t *testing.T) { func TestStartWorkerNeverInheritsTheConnectorsEnvironment(t *testing.T) { t.Setenv("CONNECTOR_CANARY_NOT_REAL", "leaked") out := filepath.Join(t.TempDir(), "env.txt") - w, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, + w, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/bin/sh", Args: []string{"-c", "env > " + out}, Env: []string{"ONLY=this"}}) require.NoError(t, err) <-w.Done() @@ -44,7 +44,7 @@ func TestStartWorkerNeverInheritsTheConnectorsEnvironment(t *testing.T) { assert.Contains(t, string(data), "ONLY=this") // A nil Env is not "inherit". - w, err = StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, + w, err = StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/bin/sh", Args: []string{"-c", "env > " + out}}) require.NoError(t, err) <-w.Done() @@ -61,11 +61,11 @@ func (refusingLauncher) Launch(context.Context, LaunchRequest) (Launched, error) func (refusingLauncher) Receipts(context.Context, string) ([]Receipt, error) { return nil, nil } func TestAStartThatRanNothingIsErrNotStarted(t *testing.T) { - _, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, Command{Path: "/nonexistent/claude-not-here"}) + _, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/nonexistent/claude-not-here"}) assert.ErrorIs(t, err, ErrNotStarted) - _, err = StartWorker(context.Background(), refusingLauncher{}, Scope{WorkDir: t.TempDir()}, Command{Path: "/bin/true"}) + _, err = StartWorker(context.Background(), SessionConfig{Launcher: refusingLauncher{}, Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/bin/true"}) assert.ErrorIs(t, err, ErrNotStarted) - _, err = StartWorker(context.Background(), nil, Scope{}, Command{Path: "/bin/true"}) + _, err = StartWorker(context.Background(), SessionConfig{}, Command{Path: "/bin/true"}) assert.ErrorIs(t, err, ErrNotStarted, "the direct launcher needs the record's directory") } @@ -76,7 +76,7 @@ func alive(pid int) bool { return syscall.Kill(pid, 0) == nil } func startWithChild(t *testing.T) (*Worker, int) { t.Helper() pidFile := filepath.Join(t.TempDir(), "child") - w, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, + w, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/bin/sh", Args: []string{"-c", "sleep 300 & echo $! > " + pidFile + "; wait"}, Env: []string{"PATH=/bin:/usr/bin"}}) require.NoError(t, err) var child int @@ -128,7 +128,7 @@ func TestTerminateReturnsWhenADescendantLeftTheGroupHoldingTheOutput(t *testing. } pidFile := filepath.Join(t.TempDir(), "escaped") script := "import os,sys,time\nif os.fork()==0:\n os.setsid()\n open(sys.argv[1],'w').write(str(os.getpid()))\n time.sleep(300)\nelse:\n time.sleep(300)\n" - w, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, + w, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: python, Args: []string{"-c", script, pidFile}, Env: []string{"PATH=/bin:/usr/bin"}}) require.NoError(t, err) var escaped int @@ -227,7 +227,7 @@ func openDescriptors(t *testing.T) int { func TestWorkersDoNotLeakDescriptors(t *testing.T) { before := openDescriptors(t) for range 50 { - _, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, Command{Path: "/nonexistent/claude-not-here"}) + _, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/nonexistent/claude-not-here"}) require.ErrorIs(t, err, ErrNotStarted) } // At most: an earlier test's worker may release its pipes meanwhile, but @@ -235,7 +235,7 @@ func TestWorkersDoNotLeakDescriptors(t *testing.T) { assert.LessOrEqual(t, openDescriptors(t), before, "fifty failed starts leave no descriptor open") for range 5 { - w, err := StartWorker(context.Background(), nil, Scope{WorkDir: t.TempDir()}, Command{Path: "/bin/true", Env: []string{}}) + w, err := StartWorker(context.Background(), SessionConfig{Scope: Scope{WorkDir: t.TempDir()}}, Command{Path: "/bin/true", Env: []string{}}) require.NoError(t, err) w.Terminate(time.Second) } diff --git a/internal/connector/driver/drivertest/drivertest.go b/internal/connector/driver/drivertest/drivertest.go index c4b535bd7..8b43db983 100644 --- a/internal/connector/driver/drivertest/drivertest.go +++ b/internal/connector/driver/drivertest/drivertest.go @@ -35,7 +35,7 @@ func StartTree(t *testing.T, dir string) (*driver.Worker, int) { // The grandchild holds the working directory open and outlives its // parent, which exits at once. script := "cd " + dir + " && (sleep 300 & echo $! > " + pidFile + ") && exit 0" - worker, err := driver.StartWorker(context.Background(), nil, driver.Scope{WorkDir: dir}, + worker, err := driver.StartWorker(context.Background(), driver.SessionConfig{Scope: driver.Scope{WorkDir: dir}}, driver.Command{Path: "/bin/sh", Args: []string{"-c", script}, Env: []string{"PATH=/bin:/usr/bin"}}) if err != nil { t.Fatalf("start a worker tree: %v", err) diff --git a/internal/connector/driver/worker.go b/internal/connector/driver/worker.go index f95c7381d..cafe23371 100644 --- a/internal/connector/driver/worker.go +++ b/internal/connector/driver/worker.go @@ -188,14 +188,22 @@ type Worker struct { releaseOnce sync.Once } -// StartWorker launches cmd through launcher, in scope, as a new process group. -// An error wrapping ErrNotStarted means no process exists; StartWorker returns -// no other error. -func StartWorker(ctx context.Context, launcher Launcher, scope Scope, cmd Command) (*Worker, error) { +// StartWorker launches cmd through cfg's launcher, in cfg's scope, as a new +// process group. An error wrapping ErrNotStarted means no process exists; +// StartWorker returns no other error. +// +// The whole of cfg is taken rather than its launcher and scope so that the +// announcement of the worker (cfg.Started, driver invariant 7) is made here, +// once, for every driver: the connector cannot arm the task token's socket +// until it knows the process group, and a driver that announced it only when +// its handshake returned would deadlock an agent whose handshake waits on +// its MCP servers. +func StartWorker(ctx context.Context, cfg SessionConfig, cmd Command) (*Worker, error) { + launcher := cfg.Launcher if launcher == nil { launcher = DirectLauncher{} } - launched, err := launcher.Launch(ctx, LaunchRequest{Scope: scope, Command: cmd}) + launched, err := launcher.Launch(ctx, LaunchRequest{Scope: cfg.Scope, Command: cmd}) if err != nil { return nil, fmt.Errorf("%w: launcher: %w", ErrNotStarted, err) } @@ -267,6 +275,13 @@ func StartWorker(ctx context.Context, launcher Launcher, scope Scope, cmd Comman w.exit = exitOf(ec, err) close(w.done) }() + // The worker exists, and nothing has been asked of it yet: this is the + // earliest the connector can be told its process group, and the latest it + // can be told without a handshake that waits on the token socket waiting + // on itself (driver invariant 7). + if cfg.Started != nil { + cfg.Started(w.process) + } return w, nil } diff --git a/internal/connector/driver/worker_other.go b/internal/connector/driver/worker_other.go index 4ac9ca54f..df6f39846 100644 --- a/internal/connector/driver/worker_other.go +++ b/internal/connector/driver/worker_other.go @@ -15,7 +15,7 @@ var errUnsupported = errors.New("driver: workers run on Unix only (process group type Worker struct{} // StartWorker refuses off Unix; nothing is started. -func StartWorker(context.Context, Launcher, Scope, Command) (*Worker, error) { +func StartWorker(context.Context, SessionConfig, Command) (*Worker, error) { return nil, errors.Join(ErrNotStarted, errUnsupported) } diff --git a/internal/connector/recovery_acp_test.go b/internal/connector/recovery_acp_test.go index 3a0c068d2..e85fa0fca 100644 --- a/internal/connector/recovery_acp_test.go +++ b/internal/connector/recovery_acp_test.go @@ -93,19 +93,21 @@ func fakeACPAdapter(w *fakeWorker) int { // from the wire, so the adapter goes on answering while its server // starts. // - // It matters which side of the handshake that is. The connector arms the - // socket for the worker's process group only once Driver.NewSession has - // returned (Dispatcher.dispatch, TokenSocket.AllowGroup on - // session.Process()), and for this driver the whole handshake — the - // adapter's own MCP read-back turn included — runs inside NewSession. A - // connection that arrives before the socket is armed waits in the - // listener's backlog, which is what that backlog is for, so arriving - // early is fine; waiting for the token before answering is not. An - // adapter whose handshake cannot finish until its MCP servers have - // connected would wait on a socket the connector cannot arm until that - // handshake finishes, and both sides would sit there until the bridge's - // 30-second dial or the driver's 2-minute handshake ran out. Nothing in - // the connector prevents that; it is the adapters that do not do it. + // It no longer matters which side of the handshake that is, and it used + // to. The connector armed the socket for the worker's process group only + // once Driver.NewSession had returned, and for this driver the whole + // handshake — the adapter's own MCP read-back turn included — runs inside + // NewSession, so an adapter that would not finish its handshake until its + // MCP servers had connected waited on a socket the connector could not + // arm until that handshake finished: both sides sat there until the + // bridge's 30-second dial or the driver's 2-minute handshake ran out. It + // is armed on the worker's process now, as soon as that process exists + // (driver invariant 7), so binding at session/new is served rather than + // deadlocked; the dispatcher's own tests hold that ordering + // (dispatcher_arming_test.go). This fake still binds where the real + // adapters do — as the handshake ends — because what this row is for is + // recovery, and the recovery it proves should be the one the adapters + // actually produce. starting := false startBind := func() { once.Do(func() { diff --git a/internal/connector/tokensocket.go b/internal/connector/tokensocket.go index b718eb29c..dccbc38fd 100644 --- a/internal/connector/tokensocket.go +++ b/internal/connector/tokensocket.go @@ -89,8 +89,23 @@ func socketDescriptor(fd uintptr) (int, bool) { // DefaultTokenWindow is how long a task token's socket waits for the worker's // MCP server once the worker exists. It covers an agent's start-up, not a // task's life, and it does not start until AllowGroup names the worker: a -// launcher or a handshake that takes its time must not spend the window of -// the worker it is still starting (card 23's review). The socket waits the +// launcher that takes its time must not spend the window of the worker it has +// not started yet (card 23's review). +// +// What the window covers moved with the arming. The connector used to name +// the worker only once Driver.NewSession had returned, so a driver's own +// handshake ran outside the window; it now names it as soon as the worker's +// process exists (driver invariant 7), because a handshake that waits on its +// MCP servers cannot otherwise be given the token those servers dial for. So +// the handshake is inside the window now, and the window means what this +// sentence has always said it means: once the WORKER exists, not once its +// session is ready. That is no smaller a window for the case it is for — an +// agent starts its MCP servers during its handshake, so their connection is +// already waiting in the listener's backlog and is accepted the moment the +// socket is armed — and it is deliberately not stretched to cover both: the +// socket is armed for the worker's whole process group, which is the agent's +// own tree, so a longer armed window is a longer window in which the agent's +// tools could ask for the token instead of its MCP server. The socket waits the // same window for the worker to be named at all, so nothing waits forever. const DefaultTokenWindow = 2 * time.Minute