diff --git a/internal/connector/recovery_claude_test.go b/internal/connector/recovery_claude_test.go new file mode 100644 index 000000000..d8f26d6f0 --- /dev/null +++ b/internal/connector/recovery_claude_test.go @@ -0,0 +1,136 @@ +//go:build unix + +package connector + +import ( + "bufio" + "context" + "encoding/json" + "os" + "slices" + "time" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" + "github.com/basecamp/basecamp-cli/internal/connector/driver/claude" +) + +// The Claude Code spawn driver's row: `claude -p` speaking stream-json. +func init() { + registerHarnessDriver(harnessDriver{ + Name: claude.Name, + New: func(agent string) driver.Driver { + return claude.New(claude.Options{Binary: agent, CloseGrace: 5 * time.Second, Lookup: func(string) (string, bool) { return "", false }}) + }, + Agent: fakeClaude, + }) + if os.Getenv(harnessRealEnv) != "" { + // The real Claude Code on PATH, for TestRecoveryAgainstRealAgents. + registerHarnessDriver(harnessDriver{ + Name: claude.Name + "-real", + Real: true, + New: func(string) driver.Driver { return claude.New(claude.Options{}) }, + }) + } +} + +// fakeClaude is `claude -p --input-format stream-json --output-format +// stream-json`: one process per session, a user message per prompt, the init +// message before the first result, and a result per turn. +func fakeClaude(w *fakeWorker) int { + args := os.Args[1:] + flag := func(name string) string { + i := slices.Index(args, name) + if i < 0 || i+1 >= len(args) { + return "" + } + return args[i+1] + } + // The server declaration is read before the init message: the driver + // removes the file once the agent reports its servers started. + var config struct { + MCPServers map[string]struct { + Command string `json:"command"` + Args []string `json:"args"` + Env map[string]string `json:"env"` + } `json:"mcpServers"` + } + data, err := os.ReadFile(flag("--mcp-config")) + if err != nil { + return 10 + } + if err := json.Unmarshal(data, &config); err != nil { + return 11 + } + names := make([]string, 0, len(config.MCPServers)) + for name, s := range config.MCPServers { + names = append(names, name) + if name == MCPServerName { + if err := w.Bind(context.Background(), driver.MCPServer{Name: name, Command: s.Command, Args: s.Args, Env: s.Env}); err != nil { + return 12 + } + } + } + + sessionID := flag("--session-id") + if sessionID == "" { + sessionID = flag("--resume") + } + mode := flag("--permission-mode") + badMode := w.BadMode() + if badMode { + mode = "bypassPermissions" + } + + out := bufio.NewWriter(os.Stdout) + emit := func(v any) { + data, _ := json.Marshal(v) + _, _ = out.Write(append(data, '\n')) + _ = out.Flush() + } + in := bufio.NewScanner(os.Stdin) + in.Buffer(make([]byte, 64<<10), 16<<20) + inited := false + for in.Scan() { + var msg struct { + Type string `json:"type"` + Message struct { + Content string `json:"content"` + } `json:"message"` + } + if json.Unmarshal(in.Bytes(), &msg) != nil { + continue + } + switch msg.Type { + case "control_request": + emit(map[string]any{"type": "result", "subtype": "error_during_execution", "is_error": true, "session_id": sessionID}) + continue + case "user": + default: + continue + } + if !inited { + inited = true + servers := make([]map[string]string, 0, len(names)) + for _, name := range names { + servers = append(servers, map[string]string{"name": name, "status": "connected"}) + } + emit(map[string]any{"type": "system", "subtype": "init", "session_id": sessionID, "permissionMode": mode, "mcp_servers": servers}) + if badMode { + // It reported the wrong mode and waits to be ended. Whether a + // real agent would already have acted is exactly what the + // connector cannot know; this one acting would only race the + // driver's kill. + w.log(0, 0, "bad-mode") + time.Sleep(2 * time.Minute) + return 9 + } + } + if err := w.Turn(context.Background(), msg.Message.Content); err != nil { + emit(map[string]any{"type": "result", "subtype": "error_during_execution", "is_error": true, "session_id": sessionID}) + continue + } + emit(map[string]any{"type": "result", "subtype": "success", "stop_reason": "end_turn", "is_error": false, "session_id": sessionID, + "usage": map[string]any{"input_tokens": 1, "output_tokens": 1}}) + } + return 0 +} diff --git a/internal/connector/recovery_connector_test.go b/internal/connector/recovery_connector_test.go new file mode 100644 index 000000000..387892592 --- /dev/null +++ b/internal/connector/recovery_connector_test.go @@ -0,0 +1,627 @@ +//go:build unix + +package connector + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed/feedtest" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/driver" + "github.com/basecamp/basecamp-cli/internal/connector/ndjson" +) + +// The connector process the recovery harness starts and kills: its kill points, +// its composition, and the ledger predicates a surviving run stops at. + +// killSpec is a run's kill point: "", or "#" to kill at the +// n-th time the point is reached rather than the first. Line points are +// "line::". +type killSpec struct { + point string + nth int + + mu sync.Mutex + count int +} + +func parseKill(raw string) *killSpec { + if raw == "" { + return &killSpec{} + } + k := &killSpec{point: raw, nth: 1} + if i := strings.LastIndex(raw, "#"); i > 0 { + if n, err := strconv.Atoi(raw[i+1:]); err == nil { + k.point, k.nth = raw[:i], n + } + } + return k +} + +// at reports whether this is the time point is to kill. +func (k *killSpec) at(point string) bool { + if k == nil || k.point != point { + return false + } + k.mu.Lock() + defer k.mu.Unlock() + k.count++ + return k.count == k.nth +} + +// die is SIGKILL, the one death nothing in the process can intercept. +func die() { + _ = syscall.Kill(os.Getpid(), syscall.SIGKILL) + select {} +} + +// killingLines is the connector's stdout: every line is kept for the parent, +// and the named line is the last thing the process does. +type killingLines struct { + mu sync.Mutex + f *os.File + kill *killSpec +} + +func (w *killingLines) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + n, err := w.f.Write(p) + if err != nil { + return n, err + } + var line struct { + Type string `json:"type"` + State string `json:"state"` + } + if json.Unmarshal(p, &line) == nil { + if line.Type == "" { + line.Type = "pointer" + } + if w.kill.at("line:" + line.Type + ":" + line.State) { + die() + } + } + return n, nil +} + +// harnessLine is one connector stdout line, as the parent reads it back. +type harnessLine struct { + Type string `json:"type"` + State string `json:"state"` + EventID int64 `json:"event_id"` + TaskID int64 `json:"task_id"` + AttemptID string `json:"attempt_id"` + EventIDs []int64 `json:"event_ids"` + StopReason string `json:"stop_reason"` + Kind string `json:"kind"` + IntentID int64 `json:"intent_id"` +} + +func (h *harness) lines() []harnessLine { + h.t.Helper() + var out []harnessLine + require.NoError(h.t, readJSONLines(filepath.Join(h.dir, linesFile), func(line []byte) error { + var l harnessLine + if err := json.Unmarshal(line, &l); err != nil { + return err + } + out = append(out, l) + return nil + })) + return out +} + +// ---- the connector process ---- + +// TestRecoveryConnector is not a test: it is the connector the harness starts +// and kills. +func TestRecoveryConnector(t *testing.T) { + if os.Getenv(harnessConnectorEnv) == "" { + t.Skip("started by the recovery harness") + } + dir := os.Getenv(harnessDirEnv) + if err := runHarnessConnector(dir); err != nil { + t.Fatal(err) + } +} + +func runHarnessConnector(dir string) error { + sc, err := readScenario(dir) + if err != nil { + return err + } + d, ok := harnessDriverNamed(sc.Driver) + if !ok { + return fmt.Errorf("no driver %q registered", sc.Driver) + } + kill := parseKill(os.Getenv(harnessKillEnv)) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + stateDir := os.Getenv(harnessStateEnv) + if stateDir == "" { + stateDir = dir + } + shadow := os.Getenv(harnessShadowEnv) == "true" + + // One connector per account and agent, as the run command takes it. + lock, err := AcquireInstanceLock(stateDir, harnessAccount, harnessAgent, time.Now()) + if err != nil { + return err + } + defer func() { _ = lock.Release() }() + ledger, err := OpenLedger(filepath.Join(stateDir, LedgerFile)) + if err != nil { + return err + } + defer func() { _ = ledger.Close() }() + // The guard is an hour out unless a scenario asks for it: a test that is + // not about the guard must not have one fire in the middle of it. + guardDelay := sc.GuardDelay + if guardDelay <= 0 { + guardDelay = time.Hour + } + hooks := LifecycleHooks(ledger, LifecycleOptions{GuardDelay: guardDelay}) + // Every task token this connector mints, kept for the parent's + // credential check — which then covers a real agent's run as well as a + // fake worker's, and a run whose worker never took its token. + launched := hooks.TaskLaunched + hooks.TaskLaunched = func(ctx context.Context, tx Tx, launch Launch) error { + if launched != nil { + if err := launched(ctx, tx, launch); err != nil { + return err + } + } + return recordTaskToken(dir, launch.AttemptID, launch.Token) + } + ended := hooks.AttemptEnded + hooks.AttemptEnded = func(ctx context.Context, tx Tx, s Settlement) error { + if err := ended(ctx, tx, s); err != nil { + return err + } + if kill.at("tx:attempt-ended") { + die() + } + return nil + } + verdict := hooks.VerdictCommitted + hooks.VerdictCommitted = func(ctx context.Context, tx Tx, v CommittedVerdict) error { + if err := verdict(ctx, tx, v); err != nil { + return err + } + if kill.at("tx:verdict") { + die() + } + return nil + } + if !shadow { + // A shadow run posts nothing, so it writes no intents. + ledger.SetHooks(hooks) + } + + out, err := os.OpenFile(filepath.Join(dir, linesFile), os.O_WRONLY|os.O_APPEND, 0o600) + if err != nil { + return err + } + defer out.Close() + lines := ndjson.NewWriter(&killingLines{f: out, kill: kill}) + + warn, pause := sc.QueueWarn, sc.QueuePause + if warn <= 0 { + warn = DefaultBacklogWarn + } + if pause <= 0 { + pause = DefaultBacklogPause + } + queue, err := NewQueue(warn, pause) + if err != nil { + return err + } + transport := feedtest.NewTransport() + minter := feedtest.NewMinter() + for range 100 { + minter.ScriptTicket(eventfeed.StreamTicket{Ticket: "test-ticket-not-real", ExpiresIn: 120, URL: "wss://cable.basecamp.com/cable?ticket=test-ticket-not-real"}) + } + var filters eventfeed.Filters + if raw := os.Getenv(harnessFiltersEnv); raw != "" { + if err := json.Unmarshal([]byte(raw), &filters); err != nil { + return err + } + } + window := sc.RepairWindow + if window <= 0 { + window = time.Minute + } + intake, err := New(Options{ + Origin: harnessOrigin, AccountID: harnessAccount, ConsumerNamespace: harnessNamespace, + Filters: filters, Ledger: ledger, Queue: queue, Minter: minter, + PollsFor: pollsFor(dir, ledger, kill, os.Getenv(harnessFaultEnv)), + Lines: lines, + Logger: logger, + Transport: transport, + RepairInterval: 50 * time.Millisecond, + RepairWindow: window, + }) + if err != nil { + return err + } + intake.repairSweep = 50 * time.Millisecond + + // Two routed projects, each its own working directory, so a test can show + // the dispatcher still runs work in one while the other's is held. + work := filepath.Join(dir, "work") + routes := map[int64]admission.Route{ + harnessBucket: {Path: work, Class: "internal"}, + harnessOtherBucket: {Path: filepath.Join(dir, "work-other"), Class: "internal"}, + } + reads := storeReads{dir: dir, gate: sc.ReadGate, kill: kill} + admitter, err := admission.NewAdmitter(admission.Policy{ + AgentID: harnessAgent, + Trust: admission.Trust{Mode: admission.TrustOperator, OperatorID: harnessOperator}, + Projects: routes, + }, admission.Reads{Summaries: reads, Subscriptions: reads, Assignments: reads}) + if err != nil { + return err + } + + mcp := WorkerMCP{Command: filepath.Join(dir, "basecamp"), Profile: "agent", StateDir: stateDir} + // Generous: a loaded box (the harness runs its own tests concurrently in + // CI, and a mutation sweep runs dozens at once) must fail on what the + // ledger says, never on how long the machine took. + runFor := harnessRunFor + if d.Real { + runFor = harnessRealRunFor + // The real `basecamp mcp`, holding a token that reaches no Basecamp: + // the worker's basecamp_connect calls are real, its Basecamp calls + // fail. + mcp.Command, mcp.Env = os.Getenv(harnessRealBasecampEnv), []string{"BASECAMP_TOKEN"} + } + failures, _ := strconv.Atoi(os.Getenv(harnessSpawnFailEnv)) + working := d.New(filepath.Join(dir, "agent")) + worker := &failingSpawns{Driver: working, broken: d.New(filepath.Join(dir, "no-such-agent")), failures: failures} + dispatcher, err := NewDispatcher(DispatcherOptions{ + Ledger: ledger, Driver: worker, + Routes: func() map[int64]admission.Route { return routes }, + Concurrency: 2, + Deadline: time.Hour, + MCP: mcp, + PrivateDir: filepath.Join(dir, "sessions"), + // As the run command wires it: the agent's replies, lifecycle + // messages filtered out by the ledger. + Replies: LifecycleFilteredReplies{Lister: storePoster{dir: dir, kill: &killSpec{}}, Ledger: ledger}, + // Not in the run command, which passes none: a task works in its + // route either way. The harness's records when a directory is + // prepared and released, for the one-owner rule's assertions. + Workspaces: &harnessWorkspaces{dir: dir}, + StillRunning: DefaultStillRunning, + IsLifecycleMessage: IsLifecycleMessageIn(ledger), + Lines: lines, + Logger: logger, + Tick: 20 * time.Millisecond, + CancelGrace: 5 * time.Second, + }) + if err != nil { + return err + } + outbox, err := NewOutbox(OutboxOptions{ + Ledger: ledger, Poster: storePoster{dir: dir, kill: kill}, + Paused: ledger.Held, Lines: lines, Logger: logger, + // A sending intent a previous process left is reconciled once it is + // this old, so a restart settles it rather than waiting out the + // production minute. + Tick: 20 * time.Millisecond, ReconcileAfter: harnessReconcileAfter, + }) + if err != nil { + return err + } + + // Who is running, for a fake worker that is to kill it: written before + // anything can be dispatched, removed when this process leaves cleanly. + identity, err := json.Marshal(map[string]any{"pid": os.Getpid(), "started_at": time.Now().UTC()}) + if err != nil { + return err + } + if err := os.WriteFile(filepath.Join(dir, connectorFile), identity, 0o600); err != nil { + return err + } + defer func() { _ = os.Remove(filepath.Join(dir, connectorFile)) }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + pushed := map[int64]bool{} + if err := readJSONLines(filepath.Join(dir, liveFile), func(line []byte) error { + var ids []int64 + if err := json.Unmarshal(line, &ids); err != nil { + return err + } + for _, id := range ids { + pushed[id] = true + } + return nil + }); err != nil { + return err + } + go (&cable{transport: transport, dir: dir, served: pushed}).run(ctx) + + var ( + wg sync.WaitGroup + errMu sync.Mutex + firstErr error + ) + part := func(name string, fn func(context.Context) error) { + wg.Go(func() { + err := fn(ctx) + if ctx.Err() == nil { + // As the run command holds it: a part that stops while the + // others run, failed or not, is a connector doing half its + // job. + if err == nil { + err = errors.New("stopped on its own") + } + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("%s: %w", name, err) + } + errMu.Unlock() + } + cancel() + }) + } + // In the run command's order: status learns the connector stopped + // however it ends; the outbox settles what a previous process left and + // sends what is due before anything transitions; only then do the parts + // start. + defer func() { _ = ledger.NoteConnection(context.Background(), ConnectionStopped, "") }() + dispatching := !shadow && os.Getenv(harnessNoDispatchEnv) != "true" + posting := dispatching && os.Getenv(harnessNoOutboxEnv) != "true" + if posting { + startCtx, stopStart := context.WithTimeout(ctx, 2*time.Minute) + err := outbox.Start(startCtx) + stopStart() + if err != nil { + return err + } + } + if err := ledger.NoteConnection(ctx, ConnectionRunning, ""); err != nil { + logger.Warn("recovery harness: note connection", "error", err) + } + part("intake", intake.Run) + part("admission", func(ctx context.Context) error { + return RunAdmission(ctx, AdmissionOptions{Ledger: ledger, Queue: queue, Admitter: admitter, Lines: lines, Logger: logger}) + }) + if dispatching { + part("dispatch", dispatcher.Run) + } + if posting { + part("outbox", outbox.Run) + } + + until := os.Getenv(harnessUntilEnv) + deadline := time.Now().Add(runFor) + for ctx.Err() == nil { + if kill.point == "paused" && queue.Paused() { + die() + } + if kill.point == "get-dispatch" { + // A worker has called get_dispatch: it cancels the guard. + var n int + if err := ledger.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_events WHERE guard = 'canceled'`).Scan(&n); err == nil && n > 0 { + die() + } + } + done, err := harnessPredicate(ctx, dir, ledger, until) + if err != nil { + logger.Warn("recovery harness: predicate", "error", err) + } + if done { + logger.Info("recovery harness: the ledger reached its predicate", "until", until, "state", unsettled(ctx, ledger)) + break + } + if time.Now().After(deadline) { + return fmt.Errorf("the ledger never reached %q; %s", until, unsettled(ctx, ledger)) + } + time.Sleep(10 * time.Millisecond) + } + cancel() + wg.Wait() + flushCtx, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + if posting { + if err := outbox.Flush(flushCtx); err != nil { + return err + } + } + return firstErr +} + +// failingSpawns starts its first workers with an agent binary that does not +// exist, so the driver itself reports a start that ran nothing. +type failingSpawns struct { + driver.Driver + broken driver.Driver + mu sync.Mutex + failures int +} + +func (f *failingSpawns) NewSession(ctx context.Context, cfg driver.SessionConfig) (driver.Session, error) { + f.mu.Lock() + fail := f.failures > 0 + if fail { + f.failures-- + } + f.mu.Unlock() + if fail { + return f.broken.NewSession(ctx, cfg) + } + return f.Driver.NewSession(ctx, cfg) +} + +// harnessPredicate is the ledger state a surviving run stops at; predicates +// joined by commas must all hold. +func harnessPredicate(ctx context.Context, dir string, l *Ledger, until string) (bool, error) { + if strings.Contains(until, ",") { + for _, one := range strings.Split(until, ",") { + ok, err := harnessPredicate(ctx, dir, l, one) + if err != nil || !ok { + return false, err + } + } + return true, nil + } + switch until { + case "", "settled": + return ledgerSettled(ctx, dir, l) + case "never": + return false, nil + } + if arg, ok := strings.CutPrefix(until, "state:"); ok { + // "state:=", or "state:" for any state. + id, state, _ := strings.Cut(arg, "=") + n, err := strconv.ParseInt(id, 10, 64) + if err != nil { + return false, err + } + r, found, err := l.Get(ctx, n) + return found && (state == "" || string(r.State) == state), err + } + if until == "losses-closed" { + settled, err := ledgerSettled(ctx, dir, l) + if err != nil || !settled { + return false, err + } + open, err := l.OpenLosses(ctx) + return len(open) == 0, err + } + return false, fmt.Errorf("unknown predicate %q", until) +} + +// How long a run may take before it fails on its predicate, and how long the +// harness waits for the process itself. A real agent calls a model, so it +// gets longer; the harness's own cap is longer than either, so a run always +// fails on what the ledger says. +const ( + harnessRunFor = 2 * time.Minute + harnessRealRunFor = 5 * time.Minute + harnessRunCap = 7 * time.Minute +) + +// harnessReconcileAfter is how old a sending intent must be before it is +// reconciled: the production minute, shortened so a restart can settle one. +const harnessReconcileAfter = 200 * time.Millisecond + +// recordTaskToken writes a task's token where the parent's credential check +// reads it. The file is outside every directory that check scans. +func recordTaskToken(dir, attemptID, token string) error { + if token == "" { + return errors.New("recovery harness: a launch with no token") + } + tokens := filepath.Join(dir, tokensDir) + if err := os.MkdirAll(tokens, 0o700); err != nil { + return err + } + return os.WriteFile(filepath.Join(tokens, attemptID+".token"), []byte(token), 0o600) +} + +// harnessWorkspaces is the working directory a task gets: the route itself, +// as the run command's default does. It records every preparation and every +// release, so a test can say whether a directory was released — which the +// one-owner rule allows only once a worker's process group is gone. +type harnessWorkspaces struct{ dir string } + +type workspaceEvent struct { + Step string `json:"step"` + Route string `json:"route"` + WorkDir string `json:"work_dir"` + EventID int64 `json:"event_id,omitempty"` +} + +func (w *harnessWorkspaces) Prepare(_ context.Context, route string, eventID int64) (string, error) { + return route, appendJSONLine(filepath.Join(w.dir, workspaceFile), workspaceEvent{Step: "prepare", Route: route, WorkDir: route, EventID: eventID}) +} + +func (w *harnessWorkspaces) Finish(_ context.Context, route, workDir string) error { + return appendJSONLine(filepath.Join(w.dir, workspaceFile), workspaceEvent{Step: "finish", Route: route, WorkDir: workDir}) +} + +// unsettled says what a run that never reached its predicate was still +// holding, so a failure names it rather than the timeout alone. +func unsettled(ctx context.Context, l *Ledger) string { + var out []string + rows, err := l.db.QueryContext(ctx, `SELECT state, COUNT(*) FROM events GROUP BY state`) + if err != nil { + return "the ledger could not be read: " + err.Error() + } + for rows.Next() { + var state string + var n int + if rows.Scan(&state, &n) == nil { + out = append(out, fmt.Sprintf("%s=%d", state, n)) + } + } + _ = rows.Close() + attempts, _ := l.LiveAttempts(ctx) + intents, _ := l.Intents(ctx, IntentFilter{States: []IntentState{IntentPending, IntentSending}}) + for _, in := range intents { + out = append(out, fmt.Sprintf("intent %d %s %s not_before=%s", in.ID, in.Kind, in.State, in.NotBefore.Format(time.RFC3339))) + } + return fmt.Sprintf("records %v, live attempts %d", out, len(attempts)) +} + +// ledgerSettled is a connector with nothing left to do: every event the feed +// serves is in the ledger, nothing waits for admission or a worker, no +// attempt is live, and no due lifecycle message is unsent. +func ledgerSettled(ctx context.Context, dir string, l *Ledger) (bool, error) { + entries, err := readFeed(dir) + if err != nil { + return false, err + } + // One query for the whole feed rather than a read per event: the overflow + // scenario publishes ten thousand of them, and this runs on a timer. + repairPolls := countRepairPolls(dir) + var want, lowest, highest int64 + for _, e := range entries { + if e.FromRepairPoll > repairPolls || e.Never { + continue + } + want++ + if lowest == 0 || e.Event.ID < lowest { + lowest = e.Event.ID + } + highest = max(highest, e.Event.ID) + } + if want > 0 { + var have int64 + if err := l.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM events WHERE id BETWEEN ? AND ?`, lowest, highest).Scan(&have); err != nil { + return false, err + } + if have < want { + return false, nil + } + } + var busy int + err = l.db.QueryRowContext(ctx, ` +SELECT (SELECT COUNT(*) FROM events WHERE state IN ('seen', 'admitted', 'queued', 'dispatched') + AND NOT (state IN ('admitted', 'queued') AND EXISTS (SELECT 1 FROM hold_marker))) + + (SELECT COUNT(*) FROM attempts WHERE state <> 'ended') + + (SELECT COUNT(*) FROM outbox WHERE state = 'sending' + OR (state = 'pending' AND not_before <= ? AND NOT EXISTS (SELECT 1 FROM hold_marker)))`, l.timestamp()).Scan(&busy) + if err != nil { + return false, err + } + return busy == 0, nil +} diff --git a/internal/connector/recovery_dispatch_test.go b/internal/connector/recovery_dispatch_test.go new file mode 100644 index 000000000..08b60ea2a --- /dev/null +++ b/internal/connector/recovery_dispatch_test.go @@ -0,0 +1,756 @@ +//go:build unix + +package connector + +import ( + "context" + "database/sql" + "math" + "os" + "os/exec" + "slices" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" + "github.com/basecamp/basecamp-cli/internal/connector/driver/drivertest" +) + +// Dispatch, acknowledgement and completion, with the connector killed at every +// ledger state an event passes through on its way to a worker and back. + +// harnessAttempt is an attempts row. +type harnessAttempt struct { + ID string + TaskID int64 + State string + StopReason string + SpawnFailed bool +} + +func harnessAttempts(t *testing.T, l *Ledger) []harnessAttempt { + t.Helper() + rows, err := l.db.QueryContext(context.Background(), `SELECT id, task_id, state, COALESCE(stop_reason, ''), spawn_failed FROM attempts ORDER BY launched_at, rowid`) + require.NoError(t, err) + defer rows.Close() + var out []harnessAttempt + for rows.Next() { + var a harnessAttempt + require.NoError(t, rows.Scan(&a.ID, &a.TaskID, &a.State, &a.StopReason, &a.SpawnFailed)) + out = append(out, a) + } + require.NoError(t, rows.Err()) + return out +} + +// outcomeOf is the outcome on the event's latest task. +func outcomeOf(t *testing.T, l *Ledger, eventID int64) string { + t.Helper() + var outcome string + require.NoError(t, l.db.QueryRowContext(context.Background(), + `SELECT COALESCE(outcome, '') FROM task_events WHERE event_id = ? ORDER BY task_id DESC LIMIT 1`, eventID).Scan(&outcome)) + return outcome +} + +// handed counts the times a worker was prompted with the event, across every +// agent process of the harness. +func (h *harness) handed(eventID int64) int { + n := 0 + for _, e := range h.agentLog() { + if e.Event == eventID && e.Step == "prompt" { + n++ + } + } + return n +} + +// agentStarts counts the agent processes that started. +func (h *harness) agentStarts() int { + n := 0 + for _, e := range h.agentLog() { + if e.Step == "start" { + n++ + } + } + return n +} + +// recordedWorkers is the worker processes the ledger recorded, which is all a +// restart has to end them by. +func recordedWorkers(t *testing.T, l *Ledger) []int { + t.Helper() + rows, err := l.db.QueryContext(context.Background(), `SELECT pid FROM attempts WHERE pid IS NOT NULL AND pid > 0`) + require.NoError(t, err) + defer rows.Close() + var out []int + for rows.Next() { + var pid int + require.NoError(t, rows.Scan(&pid)) + out = append(out, pid) + } + require.NoError(t, rows.Err()) + return out +} + +// notices are the connector's completion notices naming the event. +func (h *harness) notices(eventID int64) []storedMessage { + var out []storedMessage + for _, m := range h.connectorPosts() { + if strings.Contains(m.Content, "automatic notice") && strings.Contains(m.Content, "Event "+strconv.FormatInt(eventID, 10)+":") { + out = append(out, m) + } + } + return out +} + +// processGone says pid no longer runs: it does not exist, or it is a zombie +// nobody has reaped yet. The state comes from /proc where there is one, and +// from ps elsewhere (macOS), since a zombie still answers kill(pid, 0). +func processGone(ctx context.Context, pid int) bool { + if err := syscall.Kill(pid, 0); err != nil { + return true + } + if stat, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat"); err == nil { + // The state follows the parenthesised command name. + fields := strings.Fields(string(stat[strings.LastIndexByte(string(stat), ')')+1:])) + return len(fields) > 0 && (fields[0] == "Z" || fields[0] == "X") + } + out, err := exec.CommandContext(ctx, "ps", "-o", "stat=", "-p", strconv.Itoa(pid)).Output() + state := strings.TrimSpace(string(out)) + if err != nil && state == "" { + // ps exits non-zero when the pid names no process. + return true + } + return strings.HasPrefix(state, "Z") +} + +// completedWork is a worker that does the whole job. +var completedWork = []string{"get", "ack", "reply", "complete"} + +// crashRow is one kill point in an event's life. +type crashRow struct { + name string + // kill is where the connector kills itself; empty when the plan's worker + // kills it. + kill string + // plan is the worker's script for the event's first prompt. + plan []string + + // handed is how many times a worker was given the event, in all. + handed int + // outcome is the event's outcome once recovered. + outcome Outcome + // stop is the one attempt's stop reason once recovered. + stop StopReason + // notices is how many completion notices name the event. + notices int + // indeterminate is how many lifecycle messages wait for a person. + indeterminate int + // race marks the rows that also run under the race detector. + race bool + // noDispatch kills a connector running without its dispatcher, so the + // record is left admitted rather than racing a launch. + noDispatch bool + // noOutbox kills a connector running without its outbox, so a notice + // the settlement wrote is left pending rather than racing its claim. + noOutbox bool +} + +var crashRows = []crashRow{ + {name: "seen, the read in flight", kill: "read:5001", plan: completedWork, + handed: 1, outcome: OutcomeSucceeded, stop: StopFinished}, + {name: "seen, the verdict not committed", kill: "tx:verdict", plan: completedWork, + handed: 1, outcome: OutcomeSucceeded, stop: StopFinished}, + {name: "admitted", kill: "line:event:admitted", plan: completedWork, noDispatch: true, + handed: 1, outcome: OutcomeSucceeded, stop: StopFinished}, + {name: "dispatched, attempt running before the prompt", kill: "line:dispatch:running", plan: completedWork, + handed: 0, outcome: OutcomeUnknown, stop: StopLost, notices: 1}, + {name: "exposed by get_dispatch", plan: []string{"get", "grandchild", "kill", "linger"}, + handed: 1, outcome: OutcomeUnknown, stop: StopLost, notices: 1, race: true}, + {name: "delivered by ack_dispatch", plan: []string{"get", "ack", "grandchild", "kill", "linger"}, + handed: 1, outcome: OutcomeUnknown, stop: StopLost, notices: 1}, + {name: "completed by complete_dispatch", plan: []string{"get", "ack", "reply", "complete", "grandchild", "kill", "linger"}, + handed: 1, outcome: OutcomeSucceeded, stop: StopLost}, + {name: "worker gone, settlement not committed", kill: "tx:attempt-ended", plan: completedWork, + handed: 1, outcome: OutcomeSucceeded, stop: StopLost}, + {name: "settled, completion notice due", kill: "line:dispatch:ended", plan: []string{"get", "ack", "fail"}, noOutbox: true, + handed: 1, outcome: OutcomeFailed, stop: StopFinished, notices: 1}, + {name: "completion notice sending, not posted", kill: "post-before", plan: []string{"get", "ack", "fail"}, + handed: 1, outcome: OutcomeFailed, stop: StopFinished, indeterminate: 1}, + {name: "completion notice posted, receipt not recorded", kill: "post-after", plan: []string{"get", "ack", "fail"}, + handed: 1, outcome: OutcomeFailed, stop: StopFinished, notices: 1, race: true}, +} + +// Recovery never re-runs an instruction a worker may have seen, never resends +// a lifecycle message, and leaves an intent it cannot settle indeterminate and +// visible; everything the crash interrupted before a worker existed runs once. +func TestRecoveryAtEveryLedgerState(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + for _, row := range crashRows { + t.Run(row.name, func(t *testing.T) { + raceSubset(t, row.race) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{"101#1": row.plan}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Kill: row.kill, Killed: true, NoDispatch: row.noDispatch, NoOutbox: row.noOutbox}) + if kind, ok := strings.CutPrefix(row.kill, "line:"); ok { + lines := h.lines() + require.NotEmpty(t, lines) + last := lines[len(lines)-1] + assert.Equal(t, kind, last.Type+":"+last.State, "the connector's last word was the line it was killed at") + } + var lingering []int + if slices.Contains(row.plan, "linger") { + // The crash left a worker running: it is the restart's to + // end, by the process group the ledger recorded. + lingering = recordedWorkers(t, h.ledger()) + require.NotEmpty(t, lingering, "the crash left a worker the ledger recorded") + for _, pid := range lingering { + assert.False(t, processGone(context.Background(), pid), "the worker outlived the connector, pid %d", pid) + } + } + + children := h.children() + h.run(harnessRun{}) + h.assertRecovered(row) + for _, pid := range lingering { + assert.True(t, processGone(context.Background(), pid), "the restart ended the worker the crash left, pid %d", pid) + } + // The worker's own child too: it is ended as a group, not as + // a pid. + for _, child := range children { + assert.True(t, processGone(context.Background(), child.PID), "the restart ended the worker's child, pid %d", child.PID) + } + + // A second restart finds nothing to do and sends nothing. + posts := len(h.connectorPosts()) + h.run(harnessRun{}) + h.assertRecovered(row) + assert.Len(t, h.connectorPosts(), posts, "recovery never resends a lifecycle message") + h.assertNoWorkerOutlivedItsRecord() + }) + } + }) +} + +// assertNoWorkerOutlivedItsRecord holds the one-owner rule on every attempt +// the ledger settled: the worker it recorded is not the process running under +// that pid, and its process group has no members left. A settled record with +// any of its tree still running would be work going on with nobody owning it. +func (h *harness) assertNoWorkerOutlivedItsRecord() { + t := h.t + t.Helper() + for _, a := range recordedAttempts(t, h.ledger()) { + if a.state != string(AttemptEnded) { + continue + } + owns, err := driver.OwnsWorker(a.process) + assert.False(t, owns, "attempt %s is ended, but its worker (pid %d) still runs", a.id, a.process.PID) + assert.NoError(t, err, "attempt %s is ended, but its process group %d still has members", a.id, a.process.PGID) + } +} + +type recordedAttempt struct { + id, state string + process driver.Process +} + +// recordedAttempts is every attempt whose worker the ledger recorded: pid, +// group and start time, the identity the one-owner rule acts on. +func recordedAttempts(t *testing.T, l *Ledger) []recordedAttempt { + t.Helper() + rows, err := l.db.QueryContext(context.Background(), `SELECT id, state, pid, pgid, process_started FROM attempts WHERE pid IS NOT NULL AND pid > 0`) + require.NoError(t, err) + defer rows.Close() + var out []recordedAttempt + for rows.Next() { + var ( + a recordedAttempt + started sql.NullString + ) + require.NoError(t, rows.Scan(&a.id, &a.state, &a.process.PID, &a.process.PGID, &started)) + if started.Valid { + a.process.StartedAt, err = parseStamp(started.String) + require.NoError(t, err) + } + out = append(out, a) + } + require.NoError(t, rows.Err()) + return out +} + +func (h *harness) assertRecovered(row crashRow) { + t := h.t + t.Helper() + l := h.ledger() + assert.Equal(t, row.handed, h.handed(101), "times a worker was given the event") + assert.Equal(t, StateCompleted, stateOf(t, l, 101)) + assert.Equal(t, string(row.outcome), outcomeOf(t, l, 101)) + attempts := harnessAttempts(t, l) + if assert.Len(t, attempts, 1, "no second attempt") { + assert.Equal(t, string(row.stop), attempts[0].StopReason) + } + assert.Len(t, h.notices(101), row.notices, "completion notices") + + status, err := l.Status(context.Background(), nil) + require.NoError(t, err) + assert.Len(t, status.Indeterminate, row.indeterminate, "indeterminate lifecycle messages in status") + for _, in := range status.Indeterminate { + assert.Equal(t, string(IntentCompletion), in.Kind) + } +} + +// An unknown outcome is posted as needing a person, and a person's redispatch +// runs the event again, once. +func TestRecoveryPostsUnknownAndRedispatchRunsItAgain(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{"101#1": {"get", "ack", "kill", "linger"}}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Killed: true}) + h.run(harnessRun{}) + + notices := h.notices(101) + require.Len(t, notices, 1) + assert.Contains(t, notices[0].Content, "Event 101: unknown") + assert.Contains(t, notices[0].Content, "basecamp connect redispatch 101") + assert.Equal(t, int64(5001), notices[0].RecordingID, "on the recording that asked") + + l := h.ledger() + got, err := l.Redispatch(context.Background(), 101, "operator") + require.NoError(t, err) + assert.False(t, got.Held) + h.run(harnessRun{}) + + assert.Equal(t, 2, h.handed(101), "once before the crash, once by the redispatch") + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, 101)) + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 2) + assert.NotEqual(t, attempts[0].TaskID, attempts[1].TaskID, "a redispatch is a new task") + assert.Equal(t, string(StopFinished), attempts[1].StopReason) + assert.Len(t, h.notices(101), 1, "the redispatch succeeded with a reply: no second notice") + + h.run(harnessRun{}) + assert.Equal(t, 2, h.handed(101), "a restart after the redispatch runs nothing again") + }) +} + +// A start that ran nothing is retried once, across a restart as well, and a +// second failure blocks the record; a start whose process existed is never +// retried. +func TestRecoveryRetriesASpawnErrorOnce(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + t.Run("twice in one run", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{SpawnFail: 2}) + l := h.ledger() + assertSpawnBlocked(t, h, l) + }) + t.Run("the retry after a restart runs", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{SpawnFail: 1, Kill: "line:dispatch:ended", Killed: true}) + l := h.ledger() + assert.Equal(t, StateAdmitted, stateOf(t, l, 101), "the withdrawn exposure is durable") + + h.run(harnessRun{}) + assert.Equal(t, 1, h.handed(101)) + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, 101)) + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 2) + assert.True(t, attempts[0].SpawnFailed) + assert.False(t, attempts[1].SpawnFailed) + }) + t.Run("the retry budget survives a restart", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{SpawnFail: 1, Kill: "line:dispatch:ended", Killed: true}) + h.run(harnessRun{SpawnFail: 1}) + l := h.ledger() + assertSpawnBlocked(t, h, l) + h.run(harnessRun{}) + assert.Len(t, harnessAttempts(t, l), 2, "a blocked record is not started again by a restart") + }) + t.Run("a start that failed its handshake", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{BadModeStarts: 1}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{}) + h.run(harnessRun{}) + l := h.ledger() + assert.Equal(t, StateCompleted, stateOf(t, l, 101)) + assert.Equal(t, string(OutcomeUnknown), outcomeOf(t, l, 101), "a process existed: it may have acted") + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 1, "never retried") + assert.False(t, attempts[0].SpawnFailed) + assert.Equal(t, string(StopFailed), attempts[0].StopReason) + assert.Equal(t, 1, h.agentStarts()) + assert.Len(t, h.notices(101), 1) + }) + }) +} + +func assertSpawnBlocked(t *testing.T, h *harness, l *Ledger) { + t.Helper() + record := getRecord(t, l, 101) + assert.Equal(t, StateBlocked, record.State) + assert.Equal(t, ReasonSpawnFailed, record.Reason) + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 2, "one automatic retry, no third start") + for _, a := range attempts { + assert.True(t, a.SpawnFailed) + } + assert.Zero(t, h.agentStarts(), "no worker process ever existed") + notices := h.notices(101) + require.Len(t, notices, 1) + assert.Contains(t, notices[0].Content, "could not be started") +} + +// Follow-ups and their siblings survive a task's end, a crash included: each +// event is settled on its own, an event never handed to a worker waits for a +// task of its own, and an exposed one is never run again. +func TestRecoveryFollowUpsSurviveTheirTasksEnd(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + t.Run("a follow-up arrives, the task ends", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{ + "101#1": {"get", "arrive:102", "await:102=dispatched", "ack", "reply", "complete"}, + }}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{}) + l := h.ledger() + for _, id := range []int64{101, 102} { + assert.Equal(t, 1, h.handed(id), "event %d", id) + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, id), "event %d", id) + } + assert.Empty(t, h.connectorPosts()) + }) + t.Run("the connector dies before the follow-up is handed over", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{ + "101#1": {"get", "arrive:102", "await:102=dispatched", "kill", "linger"}, + }}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Killed: true}) + h.run(harnessRun{}) + l := h.ledger() + assert.Equal(t, 1, h.handed(101)) + assert.Equal(t, string(OutcomeUnknown), outcomeOf(t, l, 101)) + assert.Equal(t, 1, h.handed(102), "the follow-up was never exposed, so it runs as a task of its own") + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, 102)) + assert.Len(t, harnessAttempts(t, l), 2) + assert.Len(t, h.notices(101), 1) + assert.Empty(t, h.notices(102)) + }) + t.Run("the connector dies with the follow-up exposed", func(t *testing.T) { + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{ + "101#1": {"get", "arrive:102", "await:102=dispatched", "ack", "reply", "complete"}, + "102#1": {"get", "kill", "linger"}, + }}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Killed: true}) + h.run(harnessRun{}) + l := h.ledger() + assert.Equal(t, 1, h.handed(101)) + assert.Equal(t, 1, h.handed(102), "exposed: never run again") + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, 101), "a reported outcome stands") + assert.Equal(t, StateCompleted, stateOf(t, l, 102)) + assert.Equal(t, string(OutcomeUnknown), outcomeOf(t, l, 102)) + assert.Empty(t, h.notices(101)) + assert.Len(t, h.notices(102), 1) + }) + }) +} + +// measuredDispatchPromptTokens is the production-sized dispatch prompt below +// counted by a real tokenizer, once: Claude Opus 5 counted it at 296 tokens — +// Claude Code's reported input usage for the prompt, minus the same session +// with a one-character prompt (2809 - 2513), on 2026-09-17, at 810 bytes. +// Other agents' tokenizers are not measured. The test cannot re-derive it; +// estimateTokens is the bound the budget is asserted on, and it has to stay +// above this. +const measuredDispatchPromptTokens = 296 + +// The dispatch prompt is measured as the worker received it, through each +// driver's wire, at production-sized ids, and at its worst case: the largest +// ids and the longest recording URL the prompt repeats. +func TestRecoveryTheDispatchPromptIsUnderBudget(t *testing.T) { + const ( + event = int64(17_099_838_500) + followUp = int64(17_099_838_501) + recording = int64(10_304_029_146) + ) + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{ + strconv.FormatInt(event, 10) + "#1": {"get", "arrive:" + strconv.FormatInt(followUp, 10), "await:" + strconv.FormatInt(followUp, 10) + "=dispatched", "ack", "reply", "complete"}, + }}) + h.publish(feedEntry{Event: todoEvent(event, recording)}) + h.run(harnessRun{}) + + prompts := map[int64]string{} + for _, e := range h.agentLog() { + if e.Step == "prompt" { + prompts[e.Event] = e.Prompt + } + } + require.Contains(t, prompts, event) + require.Contains(t, prompts, followUp) + for id, prompt := range prompts { + tokens := estimateTokens(prompt) + t.Logf("%s: prompt for event %d: %d bytes, %d tokens by the bound, budget %d", d.Name, id, len(prompt), tokens, MaxPromptTokens) + assert.Less(t, tokens, MaxPromptTokens) + if id == event { + assert.Greater(t, tokens, measuredDispatchPromptTokens, "the bound is above what a real tokenizer counted for this prompt") + } + assert.NotContains(t, prompt, "please do the thing", "no content in the prompt") + } + if out := os.Getenv("BASECAMP_RECOVERY_PROMPT_OUT"); out != "" { + require.NoError(t, os.WriteFile(out, []byte(prompts[event]), 0o600)) + } + }) + + t.Run("worst case", func(t *testing.T) { + // The most the prompt can carry: ids at the end of their range, and a + // recording URL at the longest the prompt repeats. A longer one is + // omitted whole, so it cannot be the worst case. + longest := "https://app.basecamp.com/" + strings.Repeat("9", MaxPromptURL-len("https://app.basecamp.com/")) + record := Record{ID: math.MaxInt64, Decision: Decision{Trigger: "completed", RecordingURL: longest}} + prompt := DispatchPrompt(Launch{TaskID: math.MaxInt64}, record) + require.Contains(t, prompt, longest, "the longest URL the prompt repeats") + tokens := estimateTokens(prompt) + t.Logf("worst-case dispatch prompt: %d bytes, %d tokens by the bound, budget %d", len(prompt), tokens, MaxPromptTokens) + assert.Less(t, tokens, MaxPromptTokens) + + tooLong := longest + "9" + assert.NotContains(t, DispatchPrompt(Launch{TaskID: math.MaxInt64}, + Record{ID: math.MaxInt64, Decision: Decision{Trigger: "completed", RecordingURL: tooLong}}), + tooLong, "a URL past the cap is omitted, not carried") + + followUp := FollowUpPrompt(math.MaxInt64) + assert.Less(t, estimateTokens(followUp), MaxPromptTokens) + }) +} + +// An attempt whose worker the connector cannot identify is held, not settled: +// it stays live in the ledger, its conversation and its working directory +// stay its own, and no restart runs anything for it — while the dispatcher +// goes on running work that does not need them. +// +// The crash here is at launching, just after the attempt is written and +// before the driver is asked for anything. No process exists, but the ledger +// cannot know that: from the ledger it is the same as a crash between the +// spawn and the write of the worker's pid, which is the case the rule is for. +// That window itself cannot be hit deterministically from outside. +func TestRecoveryHoldsAnAttemptItCannotIdentify(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{"101#1": completedWork}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Kill: "line:dispatch:launching", Killed: true}) + + l := h.ledger() + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 1) + assert.Equal(t, string(AttemptLaunching), attempts[0].State, "killed before the worker's process was recorded") + + // However often it restarts. Each restart runs until work in the + // other project has been dispatched and finished: proof that its + // recovery returned and its dispatcher went on, not merely that it + // logged a decision. + // A further event in the held project, on another recording, needs the + // held directory: it waits. + h.publish(feedEntry{Event: todoEvent(104, 5004)}) + for i, other := range []int64{105, 106} { + h.publish(feedEntry{Event: otherTodoEvent(other, 6001+int64(i))}) + h.run(harnessRun{Until: "state:" + strconv.FormatInt(other, 10) + "=completed", + RequireLog: "cannot be identified"}) + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, other)) + } + assert.Equal(t, StateAdmitted, stateOf(t, l, 104), "nothing new starts in the held directory") + assert.Equal(t, 0, h.handed(104)) + assert.Equal(t, StateDispatched, stateOf(t, l, 101), "the record stays live: nobody may act on it but a person") + assert.Equal(t, 0, h.handed(101), "no worker was ever given the event") + attempts = harnessAttempts(t, l) + require.Len(t, attempts, 3, "the held attempt, and one for each event in the other project") + assert.Equal(t, string(AttemptLaunching), attempts[0].State) + assert.Empty(t, attempts[0].StopReason) + assert.False(t, h.releasedDir(h.workDir()), "the held task's working directory is not released") + assert.Empty(t, h.notices(101), "an attempt that is still live has no completion to post") + }) +} + +// The guard acknowledgement is the connector's own message, and a crash around +// its post leaves it sent once or indeterminate — never twice. +func TestRecoveryTheGuardAcknowledgementIsPostedAtMostOnce(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + for _, row := range []struct { + name string + kill string + boosts int + waiting int + }{ + {name: "posted, receipt not recorded", kill: "post-after", boosts: 1}, + {name: "sending, not posted", kill: "post-before", waiting: 1}, + } { + t.Run(row.name, func(t *testing.T) { + // A worker that never calls get_dispatch is what the guard is + // for: the acknowledgement falls to the connector. + h := newHarness(t, d, harnessScenario{ + GuardDelay: 50 * time.Millisecond, + Plans: map[string][]string{"101#1": {"linger"}}, + }) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Kill: row.kill, Killed: true}) + h.run(harnessRun{}) + h.run(harnessRun{}) + + var boosts []storedMessage + for _, m := range h.connectorPosts() { + if m.Kind == MessageBoost { + boosts = append(boosts, m) + } + } + assert.Len(t, boosts, row.boosts, "guard acknowledgements posted") + for _, boost := range boosts { + assert.Equal(t, GuardAckBody, boost.Content) + assert.Equal(t, int64(5001), boost.RecordingID) + } + status, err := h.ledger().Status(context.Background(), nil) + require.NoError(t, err) + waiting := 0 + for _, in := range status.Indeterminate { + if in.Kind == string(IntentGuardAck) { + waiting++ + } + } + assert.Equal(t, row.waiting, waiting, "guard acknowledgements waiting for a person") + }) + } + }) +} + +// One owner, one release point: a worker's tree that outlives it keeps its +// attempt live, its record non-terminal and its working directory unreleased, +// through any number of restarts, because recovery holds an attempt whose +// worker it cannot verify rather than settling around it — while it goes on +// running work that does not need that directory. Once the tree is gone, the +// next restart settles the attempt and releases the directory. +func TestRecoveryAWorkersSurvivingTreeKeepsItsAttempt(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{"101#1": {"get", "grandchild", "kill", "exit:0"}}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Killed: true}) + + children := h.children() + require.Len(t, children, 1, "the worker started its grandchild") + grandchild := children[0] + l := h.ledger() + attempts := recordedAttempts(t, l) + require.Len(t, attempts, 1) + worker := attempts[0].process + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + // Exited and reaped, not a zombie: a zombie still has the start time + // the ledger recorded, and recovery would rightly take it for the + // worker. + require.NoError(t, waitFor(ctx, func() (bool, error) { return syscall.Kill(worker.PID, 0) != nil, nil }), "the worker itself exited and was reaped") + require.False(t, processGone(context.Background(), grandchild.PID), "its grandchild did not") + drivertest.RequireGroupHeld(t, worker) + + h.publish(feedEntry{Event: todoEvent(104, 5004)}) + for i, other := range []int64{105, 106} { + h.publish(feedEntry{Event: otherTodoEvent(other, 6001+int64(i))}) + h.run(harnessRun{Until: "state:" + strconv.FormatInt(other, 10) + "=completed", + RequireLog: "could not verify whether a previous worker still runs"}) + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, other), "work that does not need the held directory still runs") + assert.Equal(t, StateAdmitted, stateOf(t, l, 104), "nothing new starts in the held directory") + + assert.Equal(t, string(AttemptRunning), attemptState(t, l, attempts[0].id), "the attempt stays live while its tree runs") + assert.Equal(t, StateDispatched, stateOf(t, l, 101), "the record is not made terminal") + assert.False(t, h.releasedDir(h.workDir()), "the working directory is not released") + assert.Empty(t, h.notices(101), "an attempt that is still live has no completion to post") + assert.False(t, processGone(context.Background(), grandchild.PID), "recovery does not signal a group whose leader it cannot verify") + _, err := driver.OwnsWorker(worker) + assert.ErrorIs(t, err, driver.ErrGroupOutlivedLeader) + } + + // The tree ends; the next restart may settle and release. + killRecorded(t, grandchild) + // Reaped, not merely dead: a zombie is still a member of the group. + require.NoError(t, waitFor(ctx, func() (bool, error) { return syscall.Kill(grandchild.PID, 0) != nil, nil })) + h.run(harnessRun{}) + assert.Equal(t, string(AttemptEnded), attemptState(t, l, attempts[0].id)) + assert.Equal(t, StateCompleted, stateOf(t, l, 101)) + assert.Equal(t, string(OutcomeUnknown), outcomeOf(t, l, 101)) + assert.True(t, h.releasedDir(h.workDir()), "released once the tree is gone") + assert.Len(t, h.notices(101), 1) + assert.Equal(t, 1, h.handed(101), "and never run again") + assert.Equal(t, 1, h.handed(104), "the directory released, the waiting event runs") + h.assertNoWorkerOutlivedItsRecord() + }) +} + +// On start, the outbox settles what a previous process left sending before +// anything else runs: a notice whose receipt the crash lost is adopted before +// the restarted connector dispatches new work. +func TestRecoveryReconcilesLifecycleMessagesBeforeAnythingRuns(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + h := newHarness(t, d, harnessScenario{Plans: map[string][]string{"101#1": {"get", "ack", "fail"}}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Kill: "post-after", Killed: true}) + + // A start leaves a request younger than ReconcileAfter to land; this + // one is older, so the start must settle it. + l := h.ledger() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + require.NoError(t, waitFor(ctx, func() (bool, error) { + sending, err := l.Intents(ctx, IntentFilter{States: []IntentState{IntentSending}}) + return len(sending) == 1 && sending[0].SendingAt != nil && time.Since(*sending[0].SendingAt) > harnessReconcileAfter, err + })) + + h.publish(feedEntry{Event: otherTodoEvent(102, 6001)}) + before := len(h.lines()) + h.run(harnessRun{}) + + lines := h.lines()[before:] + reconciled, launched := -1, -1 + for i, line := range lines { + if line.Type == "outbox" && line.Kind == string(IntentCompletion) && line.State == string(IntentSent) && reconciled < 0 { + reconciled = i + } + if line.Type == "dispatch" && line.State == string(AttemptLaunching) && slices.Contains(line.EventIDs, 102) && launched < 0 { + launched = i + } + } + require.GreaterOrEqual(t, reconciled, 0, "the notice was adopted") + require.GreaterOrEqual(t, launched, 0, "the new event ran") + assert.Less(t, reconciled, launched, "the notice was settled before anything new was dispatched") + assert.Len(t, h.notices(101), 1, "adopted, not posted again") + }) +} + +func attemptState(t *testing.T, l *Ledger, id string) string { + t.Helper() + var state string + require.NoError(t, l.db.QueryRowContext(context.Background(), `SELECT state FROM attempts WHERE id = ?`, id).Scan(&state)) + return state +} + +// killRecorded SIGKILLs a process the harness recorded, only while it is still +// that process. +func killRecorded(t *testing.T, p driver.Process) { + t.Helper() + if owns, err := driver.OwnsWorker(driver.Process{PID: p.PID, PGID: p.PID, StartedAt: p.StartedAt}); err == nil && owns { + require.NoError(t, syscall.Kill(p.PID, syscall.SIGKILL)) + } +} diff --git a/internal/connector/recovery_fakes_test.go b/internal/connector/recovery_fakes_test.go new file mode 100644 index 000000000..576cd612a --- /dev/null +++ b/internal/connector/recovery_fakes_test.go @@ -0,0 +1,592 @@ +//go:build unix + +package connector + +import ( + "bufio" + "cmp" + "context" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "slices" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed/feedtest" +) + +// The recovery harness's fake account: the feed (both lanes) and Basecamp +// (messages the agent created, and admission's reads). + +// feedEntry is one event the fake account holds. +type feedEntry struct { + Event eventfeed.Event `json:"event"` + // FromRepairPoll hides the event from every repair poll before the n-th, + // the way the poll lane's safety delay withholds a committing event. + FromRepairPoll int `json:"from_repair_poll,omitempty"` + // Live is also pushed on the socket, as soon as a connection confirms. + Live bool `json:"live,omitempty"` + // Never is never served by a poll: a recording deleted before it became + // poll-visible. + Never bool `json:"never,omitempty"` +} + +// todoEvent is a to-do created by the operator that mentions the agent. A +// to-do is its own conversation, so events on one recording queue behind each +// other. +func todoEvent(id, recording int64) eventfeed.Event { + return eventfeed.Event{ + ID: id, Kind: "todo_created", EventType: "todo.created", Action: "created", + CreatedAt: time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC), + BucketID: harnessBucket, CreatorID: harnessOperator, RecordingID: recording, + } +} + +// otherTodoEvent is todoEvent in the second routed project. +func otherTodoEvent(id, recording int64) eventfeed.Event { + e := todoEvent(id, recording) + e.BucketID = harnessOtherBucket + return e +} + +// publish adds events to the fake account's feed. +func (h *harness) publish(entries ...feedEntry) { + h.t.Helper() + require.NoError(h.t, appendFeed(h.dir, entries...)) +} + +func appendFeed(dir string, entries ...feedEntry) error { + return withLockedFile(filepath.Join(dir, feedFile), func(f *os.File) error { + for _, e := range entries { + data, err := json.Marshal(e) + if err != nil { + return err + } + if _, err := f.Write(append(data, '\n')); err != nil { + return err + } + } + return nil + }) +} + +// feedCache keeps the last parse of a feed file by its size: the file is only +// ever appended to, and a burst of ten thousand events is read on every poll. +var feedCache struct { + sync.Mutex + path string + size int64 + entries []feedEntry +} + +func readFeed(dir string) ([]feedEntry, error) { + path := filepath.Join(dir, feedFile) + info, err := os.Stat(path) + if err != nil { + return nil, err + } + feedCache.Lock() + defer feedCache.Unlock() + if feedCache.path == path && feedCache.size == info.Size() { + return feedCache.entries, nil + } + var out []feedEntry + err = readJSONLines(path, func(line []byte) error { + var e feedEntry + if err := json.Unmarshal(line, &e); err != nil { + return err + } + out = append(out, e) + return nil + }) + if err != nil { + return nil, err + } + slices.SortFunc(out, func(a, b feedEntry) int { return cmp.Compare(a.Event.ID, b.Event.ID) }) + feedCache.path, feedCache.size, feedCache.entries = path, info.Size(), out + return out, nil +} + +// withLockedFile runs fn with the file open for appending under an exclusive +// flock: the connector and the fake agents write the same files. +func withLockedFile(path string, fn func(f *os.File) error) error { + f, err := os.OpenFile(path, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0o600) + if err != nil { + return err + } + defer f.Close() + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil { + return err + } + defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }() + return fn(f) +} + +func readJSONLines(path string, fn func(line []byte) error) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_SH); err != nil { + return err + } + defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }() + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64<<10), 16<<20) + for scanner.Scan() { + if len(scanner.Bytes()) == 0 { + continue + } + if err := fn(scanner.Bytes()); err != nil { + return err + } + } + return scanner.Err() +} + +func appendJSONLine(path string, v any) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + return withLockedFile(path, func(f *os.File) error { + _, err := f.Write(append(data, '\n')) + return err + }) +} + +// pollLog is one poll the connector made. +type pollLog struct { + Repair bool `json:"repair"` + Since string `json:"since,omitempty"` + Position string `json:"position,omitempty"` + Served []int64 `json:"served"` + Stalled bool `json:"stalled,omitempty"` +} + +func (h *harness) polls() []pollLog { + h.t.Helper() + var out []pollLog + require.NoError(h.t, readJSONLines(filepath.Join(h.dir, pollsFile), func(line []byte) error { + var p pollLog + if err := json.Unmarshal(line, &p); err != nil { + return err + } + out = append(out, p) + return nil + })) + return out +} + +// filePolls is the poll lane over the feed file, one per walk: the feed's +// connection or a loss's repair walk. Positions are "feed-" and +// "repair-", both meaning "after id". +type filePolls struct { + dir string + ledger *Ledger + kill *killSpec + fault string + repair bool + + mu sync.Mutex + polled bool +} + +// pollsFor hands intake a poll source per walk, telling a repair walk from +// the feed's connection by who asked. +func pollsFor(dir string, ledger *Ledger, kill *killSpec, fault string) func() eventfeed.PollSource { + return func() eventfeed.PollSource { + pcs := make([]uintptr, 32) + frames := runtime.CallersFrames(pcs[:runtime.Callers(2, pcs)]) + repair := false + for { + frame, more := frames.Next() + if strings.HasSuffix(frame.Function, ".(*Intake).runRepair") { + repair = true + } + if !more { + break + } + } + return &filePolls{dir: dir, ledger: ledger, kill: kill, fault: fault, repair: repair} + } +} + +const maxHarnessPage = 500 + +func (p *filePolls) Poll(ctx context.Context, cursor eventfeed.Cursor, _ eventfeed.Filters) (eventfeed.PollPage, error) { + if err := ctx.Err(); err != nil { + return eventfeed.PollPage{}, err + } + p.mu.Lock() + first := !p.polled + p.polled = true + p.mu.Unlock() + logPath := filepath.Join(p.dir, pollsFile) + if p.repair { + if err := p.awaitLosses(ctx); err != nil { + return eventfeed.PollPage{}, err + } + if p.kill.at("repair-poll") { + die() + } + if p.fault == "repair-stall" { + if err := appendJSONLine(logPath, pollLog{Repair: true, Since: cursor.Since, Position: cursor.Position, Stalled: true}); err != nil { + return eventfeed.PollPage{}, err + } + <-ctx.Done() + return eventfeed.PollPage{}, ctx.Err() + } + } else { + if p.kill.at("feed-poll") { + die() + } + if p.fault == "stall-catch-up" && first { + // The feed's first walk is held while the socket's burst piles up + // in the live buffer behind it, until the overflow is on disk. + if err := p.awaitLosses(ctx); err != nil { + return eventfeed.PollPage{}, err + } + } + } + entries, err := readFeed(p.dir) + if err != nil { + return eventfeed.PollPage{}, err + } + var after int64 + switch { + case strings.HasPrefix(cursor.Position, "feed-"), strings.HasPrefix(cursor.Position, "repair-"): + _, n, _ := strings.Cut(cursor.Position, "-") + after, _ = strconv.ParseInt(n, 10, 64) + case cursor.Position != "": + return eventfeed.PollPage{}, fmt.Errorf("a position this feed never issued: %q", cursor.Position) + case cursor.Since == "now": + for _, e := range entries { + after = max(after, e.Event.ID) + } + case cursor.Since != "": + after, _ = strconv.ParseInt(cursor.Since, 10, 64) + } + // The safety delay is counted in repair polls, for both walks: the n-th + // repair poll, and every poll after it, sees what it sees. + repairPolls := countRepairPolls(p.dir) + if p.repair { + repairPolls++ + } + page := eventfeed.PollPage{} + last := after + for _, e := range entries { + if e.Event.ID <= after || e.Never { + continue + } + if e.FromRepairPoll > repairPolls { + // Still inside the safety delay: withheld, and so is everything + // after it, since a page never skips a committing event. + break + } + if len(page.Events) == maxHarnessPage { + break + } + page.Events = append(page.Events, e.Event) + last = e.Event.ID + } + prefix := "feed-" + if p.repair { + prefix = "repair-" + } + page.Position = prefix + strconv.FormatInt(last, 10) + served := make([]int64, 0, len(page.Events)) + for _, e := range page.Events { + served = append(served, e.ID) + } + if err := appendJSONLine(logPath, pollLog{Repair: p.repair, Since: cursor.Since, Position: cursor.Position, Served: served}); err != nil { + return eventfeed.PollPage{}, err + } + return page, nil +} + +// awaitLosses holds a poll until the scenario's overflow losses are all on +// disk, so a kill in the walk never races the signal that records the next. +func (p *filePolls) awaitLosses(ctx context.Context) error { + sc, err := readScenario(p.dir) + if err != nil || sc.OverflowLosses == 0 { + return err + } + return waitFor(ctx, func() (bool, error) { + var n int + err := p.ledger.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM losses`).Scan(&n) + return n >= sc.OverflowLosses, err + }) +} + +func countRepairPolls(dir string) int { + n := 0 + _ = readJSONLines(filepath.Join(dir, pollsFile), func(line []byte) error { + var l pollLog + if json.Unmarshal(line, &l) == nil && l.Repair && !l.Stalled { + n++ + } + return nil + }) + return n +} + +// cable answers every connection's subscription, and serves the feed's +// live-only events on the first connection that confirms. +type cable struct { + transport *feedtest.Transport + dir string + served map[int64]bool +} + +func (c *cable) run(ctx context.Context) { + type connState struct { + welcomed bool + identifier string + } + conns := map[*feedtest.Conn]*connState{} + ticker := time.NewTicker(5 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + for _, conn := range c.transport.Conns() { + s := conns[conn] + if s == nil { + s = &connState{} + conns[conn] = s + } + if conn.Closed() { + continue + } + if !s.welcomed { + // Action Cable greets first; the subscribe follows. + conn.Serve([]byte(`{"type":"welcome"}`)) + s.welcomed = true + } + if s.identifier == "" { + for _, w := range conn.Writes() { + var command struct { + Command string `json:"command"` + Identifier string `json:"identifier"` + } + if json.Unmarshal(w, &command) == nil && command.Command == "subscribe" && command.Identifier != "" { + frame, _ := json.Marshal(map[string]string{"type": "confirm_subscription", "identifier": command.Identifier}) + conn.Serve(frame) + s.identifier = command.Identifier + break + } + } + } + if s.identifier == "" { + continue + } + entries, err := readFeed(c.dir) + if err != nil { + continue + } + var fresh []int64 + for _, e := range entries { + if !e.Live || c.served[e.Event.ID] { + continue + } + c.served[e.Event.ID] = true + conn.Serve(liveFrame(s.identifier, e.Event)) + fresh = append(fresh, e.Event.ID) + } + if len(fresh) > 0 { + // A push happens once in the world: a restarted connector's + // socket does not hear it again. + _ = appendJSONLine(filepath.Join(c.dir, liveFile), fresh) + } + } + } +} + +func liveFrame(identifier string, event eventfeed.Event) []byte { + payload, _ := json.Marshal(map[string]any{ + "id": event.ID, "kind": event.Kind, "event_type": event.EventType, "action": event.Action, + "created_at": event.CreatedAt.UTC().Format(time.RFC3339), "bucket_id": event.BucketID, + "creator_id": event.CreatorID, "performed_by_id": nil, "actor_type": "person", + "recording_id": event.RecordingID, "visible_to_clients": true, + }) + id, _ := json.Marshal(identifier) + frame, _ := json.Marshal(map[string]json.RawMessage{"identifier": id, "message": payload}) + return frame +} + +// ---- the fake Basecamp ---- + +// storedMessage is a boost, comment or chat line the agent created: by the +// connector's outbox, or by a worker. +type storedMessage struct { + ID int64 `json:"id"` + Kind MessageKind `json:"kind"` + BucketID int64 `json:"bucket_id"` + RecordingID int64 `json:"recording_id"` + Content string `json:"content"` + // By is "connector" or "worker". + By string `json:"by"` + At time.Time `json:"at"` +} + +func postMessage(dir string, m storedMessage) (int64, error) { + var id int64 + err := withLockedFile(filepath.Join(dir, storeFile), func(f *os.File) error { + n := 0 + if _, err := f.Seek(0, io.SeekStart); err != nil { + return err + } + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64<<10), 16<<20) + for scanner.Scan() { + n++ + } + if err := scanner.Err(); err != nil { + return err + } + id = 7_000_000 + int64(n) + 1 + m.ID, m.At = id, time.Now().UTC() + data, err := json.Marshal(m) + if err != nil { + return err + } + _, err = f.Write(append(data, '\n')) + return err + }) + return id, err +} + +func storedMessages(dir string) ([]storedMessage, error) { + var out []storedMessage + err := readJSONLines(filepath.Join(dir, storeFile), func(line []byte) error { + var m storedMessage + if err := json.Unmarshal(line, &m); err != nil { + return err + } + out = append(out, m) + return nil + }) + return out, err +} + +func (h *harness) messages() []storedMessage { + h.t.Helper() + out, err := storedMessages(h.dir) + require.NoError(h.t, err) + return out +} + +// connectorPosts are the lifecycle messages the connector posted. +func (h *harness) connectorPosts() []storedMessage { + h.t.Helper() + var out []storedMessage + for _, m := range h.messages() { + if m.By == "connector" { + out = append(out, m) + } + } + return out +} + +// storePoster is the outbox's Basecamp. +type storePoster struct { + dir string + kill *killSpec +} + +func (p storePoster) Post(ctx context.Context, dest Destination, body string) (int64, error) { + if p.kill.at("post-before") { + die() + } + id, err := postMessage(p.dir, storedMessage{Kind: dest.Kind, BucketID: dest.BucketID, RecordingID: dest.RecordingID, Content: body, By: "connector"}) + if err != nil { + return 0, err + } + if p.kill.at("post-after") { + die() + } + return id, nil +} + +func (p storePoster) List(_ context.Context, dest Destination, since time.Time) ([]PostedMessage, error) { + all, err := storedMessages(p.dir) + if err != nil { + return nil, err + } + var out []PostedMessage + for _, m := range all { + if m.Kind == dest.Kind && m.RecordingID == dest.RecordingID && !m.At.Before(since) { + out = append(out, PostedMessage{ID: m.ID, CreatedAt: m.At, Content: m.Content}) + } + } + return out, nil +} + +// storeReads answers admission: every recording is a to-do the operator wrote +// that mentions the agent. +type storeReads struct { + dir string + gate []int64 + kill *killSpec +} + +func (r storeReads) Summarize(ctx context.Context, ref basecamp.RecordingRef) (*basecamp.RecordingSummary, error) { + if r.kill.at("read:" + strconv.FormatInt(ref.RecordingID, 10)) { + die() + } + if slices.Contains(r.gate, ref.RecordingID) { + waiting := filepath.Join(r.dir, "read-waiting-"+strconv.FormatInt(ref.RecordingID, 10)) + release := filepath.Join(r.dir, "read-release-"+strconv.FormatInt(ref.RecordingID, 10)) + _ = os.WriteFile(waiting, nil, 0o600) + if err := awaitFile(ctx, release); err != nil { + return nil, err + } + } + id := strconv.FormatInt(ref.RecordingID, 10) + return &basecamp.RecordingSummary{ + ID: ref.RecordingID, Status: "active", Type: "Todo", Title: "To-do " + id, + AppURL: "https://app.basecamp.com/" + harnessAccount + "/buckets/" + strconv.FormatInt(ref.BucketID, 10) + "/todos/" + id, + Bucket: &basecamp.Bucket{ID: ref.BucketID}, + Creator: &basecamp.Person{ID: harnessOperator}, + Content: mentionMarkup(harnessAgent) + " please do the thing", + MentionedPersonIDs: []int64{harnessAgent}, + UpdatedAt: time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC), + }, nil +} + +func (storeReads) Subscribed(context.Context, int64) (bool, error) { return false, nil } + +func (storeReads) AddedPersonIDs(context.Context, int64, int64) ([]int64, bool, error) { + return nil, false, nil +} + +// awaitFile waits for path to exist. It is how a process waits on a step +// another process takes. +func awaitFile(ctx context.Context, path string) error { + for { + if _, err := os.Stat(path); err == nil { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(5 * time.Millisecond): + } + } +} diff --git a/internal/connector/recovery_harness_test.go b/internal/connector/recovery_harness_test.go new file mode 100644 index 000000000..2d09f8226 --- /dev/null +++ b/internal/connector/recovery_harness_test.go @@ -0,0 +1,818 @@ +//go:build unix + +package connector + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" + "github.com/basecamp/basecamp-cli/internal/connector/driver/drivertest" +) + +// The integrated recovery harness (plan step 22). +// +// The connector under test is a real process: this test binary, started as +// TestRecoveryConnector, composed the way the run command composes it — +// intake on the feed's run loop, admission, the dispatcher with a real driver, +// and the outbox with the lifecycle hooks — over a ledger file. A test kills +// that process with SIGKILL at an injected point, starts it again over the +// same ledger, lets it run until the ledger is settled, and asserts on the +// ledger and on what reached the fake Basecamp. +// +// Nothing in the connector's own code knows about the harness, and the harness +// adds no seam to it. The kill points are: +// +// - after a commit: the connector's own stdout lines are written after the +// transaction they report, and the harness's line writer kills the process +// once it has written the line named; +// - inside a transaction: a ledger hook that kills before returning, so the +// transaction never commits; +// - in Basecamp: the fake poster kills before or after the message exists, +// the fake reads kill mid-read, the fake feed kills in a repair walk; +// - in the worker: the fake agent kills its parent, the connector, between +// get_dispatch, ack_dispatch and complete_dispatch; +// - in shadow promote and import: the named steps their own crash tests +// already kill at (TestCrashHelper). +// +// # Drivers +// +// Each driver the connector can start workers with registers a row +// (registerHarnessDriver): how to build the driver with a fake agent as its +// binary, and the fake agent's side of the driver's wire. The fake agent is +// this test binary behind a small exec wrapper, so the dispatcher's +// environment allowlist is never widened for the harness. Whatever the wire, +// the fake agent's work is the same fakeWorker: it binds to its task through +// the MCP server declaration the driver handed it (the state directory and the +// task token) and calls the ledger exactly as `basecamp mcp --connect-state` +// does. Every dispatch test runs once per registered driver. +// +// # What the harness does not cover +// +// A driver row builds its driver directly, where the run command goes through +// spawn.New(connect.json's worker): the registry that maps a worker name to a +// driver is that package's own test's. The fake agent is the driver's binary, +// which is what the registry would otherwise decide. +// +// # Synchronization +// +// No test sleeps for an outcome. The connector runs until a predicate over the +// ledger holds; the fake agent waits on ledger state; the parent waits on +// process exit. Every wait has a deadline that fails the test rather than +// hanging it. + +// harnessDriver is one row of the driver table. +type harnessDriver struct { + // Name is the row's name in test names. + Name string + // New builds the driver under test with agent as its agent executable. + New func(agent string) driver.Driver + // Agent is the fake agent: it speaks the driver's wire on stdin and + // stdout, binds w to the MCP server declaration it was given, calls + // w.Turn for each prompt, and returns the process's exit code. + Agent func(w *fakeWorker) int + // Real rows start the real agent binary with the real `basecamp mcp`: + // they run only in TestRecoveryAgainstRealAgents, opted into locally. + Real bool +} + +var harnessDrivers []harnessDriver + +// registerHarnessDriver adds a driver row. Call it from an init function in +// the driver's own recovery__test.go. +func registerHarnessDriver(d harnessDriver) { + for _, have := range harnessDrivers { + if have.Name == d.Name { + panic("recovery harness: driver " + d.Name + " registered twice") + } + } + harnessDrivers = append(harnessDrivers, d) +} + +func harnessDriverNamed(name string) (harnessDriver, bool) { + for _, d := range harnessDrivers { + if d.Name == name { + return d, true + } + } + return harnessDriver{}, false +} + +// forEachDriver runs fn as a subtest per registered driver. +func forEachDriver(t *testing.T, fn func(t *testing.T, d harnessDriver)) { + t.Helper() + require.NotEmpty(t, harnessDrivers, "no driver registered with the recovery harness") + for _, d := range harnessDrivers { + if d.Real { + continue + } + t.Run(d.Name, func(t *testing.T) { + if testing.Short() { + t.Skip("starts processes") + } + fn(t, d) + }) + } +} + +// raceSubset skips a harness case under the race detector unless it is one of +// the representative few. A race-instrumented test binary takes over a second +// to start, and the harness starts one per connector run and per worker, so +// the whole table runs in the ordinary test job and the race job runs enough +// of it to race-check the composed connector across a kill and a restart. +func raceSubset(t *testing.T, representative bool) { + t.Helper() + if harnessUnderRace && !representative { + t.Skip("under -race the recovery harness runs its representative cases; the full table runs without it") + } +} + +// Environment of the harness's processes. +const ( + harnessConnectorEnv = "BASECAMP_RECOVERY_CONNECTOR" + harnessAgentEnv = "BASECAMP_RECOVERY_AGENT" + harnessDirEnv = "BASECAMP_RECOVERY_DIR" + harnessKillEnv = "BASECAMP_RECOVERY_KILL" + harnessUntilEnv = "BASECAMP_RECOVERY_UNTIL" + harnessSpawnFailEnv = "BASECAMP_RECOVERY_SPAWN_FAIL" + harnessFiltersEnv = "BASECAMP_RECOVERY_FILTERS" + harnessStateEnv = "BASECAMP_RECOVERY_STATE" + harnessShadowEnv = "BASECAMP_RECOVERY_SHADOW" + harnessFaultEnv = "BASECAMP_RECOVERY_FAULT" + harnessNoDispatchEnv = "BASECAMP_RECOVERY_NO_DISPATCH" + harnessNoOutboxEnv = "BASECAMP_RECOVERY_NO_OUTBOX" + harnessScanEnv = "BASECAMP_RECOVERY_SECRET_SCAN" + // harnessRealEnv opts into the run against the real agent binaries, and + // harnessRealBasecampEnv names the basecamp binary built from this tree + // whose `mcp` the real workers start. + harnessRealEnv = "BASECAMP_RECOVERY_REAL_AGENTS" + harnessRealBasecampEnv = "BASECAMP_RECOVERY_BASECAMP" +) + +// The test binary doubles as a fake agent: started through the wrapper a +// harness writes, it speaks the wire of the driver it names and exits. +func TestMain(m *testing.M) { + if name := os.Getenv(harnessAgentEnv); name != "" { + os.Exit(runFakeAgent(name)) + } + if os.Getenv(harnessScanEnv) != "" { + os.Exit(runSecretScan(os.Args[1:])) + } + os.Exit(m.Run()) +} + +func runFakeAgent(name string) int { + d, ok := harnessDriverNamed(name) + if !ok { + fmt.Fprintln(os.Stderr, "recovery harness: no fake agent for driver", name) + return 97 + } + w, err := newFakeWorker(os.Getenv(harnessDirEnv)) + if err != nil { + fmt.Fprintln(os.Stderr, "recovery harness:", err) + return 98 + } + defer w.close() + return d.Agent(w) +} + +// Scenario constants: one account, one agent, one operator, one routed project. +const ( + harnessAccount = "2914079" + harnessAgent = adapterAgentID + harnessOperator = adapterOperatorID + harnessBucket = adapterBucketID + // harnessOtherBucket is a second routed project with its own directory. + harnessOtherBucket = int64(48929974) + harnessOrigin = "https://3.basecampapi.com" + harnessNamespace = "basecamp-connect-recovery" +) + +// harnessScenario is what every process of one harness reads: the connector, +// the fake agent and the parent. +type harnessScenario struct { + Driver string `json:"driver"` + // Plans script the fake worker per event and per time it was prompted + // with that event: "#", n from 1. A missing plan is the + // ordinary worker: get, ack, reply, complete. + Plans map[string][]string `json:"plans"` + // BadModeStarts is how many agent processes report a permission mode + // other than the one asked for. + BadModeStarts int `json:"bad_mode_starts"` + // QueueWarn and QueuePause size the backlog; the defaults when zero. + QueueWarn int `json:"queue_warn"` + QueuePause int `json:"queue_pause"` + // ReadGate names recordings whose admission read waits for the parent. + ReadGate []int64 `json:"read_gate"` + // GuardDelay is how long a worker has to call get_dispatch before the + // guard acknowledges; an hour when zero, so no guard fires in a test that + // is not about it. + GuardDelay time.Duration `json:"guard_delay"` + // RepairWindow is the loss window; a minute when zero. + RepairWindow time.Duration `json:"repair_window"` + // OverflowLosses is how many losses the scenario's overflow records: the + // feed's catch-up and the repair walks wait for all of them. + OverflowLosses int `json:"overflow_losses"` +} + +// harness is one scenario's directory: the connector's state directory, the +// fake Basecamp, the feed, and every process's log. +type harness struct { + t *testing.T + dir string + // watching is the parent's watch for each task token, for the run in + // flight; watchStop ends it. + watchMu sync.Mutex + watching map[string]func() []string + watchStop chan struct{} + // state is the connector's state directory, under this harness's own + // XDG_STATE_HOME and named as the connector names it, so a worker's MCP + // server resolves it exactly as `basecamp mcp --connect-state` does. + state string + agent string + driver harnessDriver + sc harnessScenario +} + +func newHarness(t *testing.T, d harnessDriver, sc harnessScenario) *harness { + t.Helper() + // Not t.TempDir: its name carries the test's, and the attempt's token + // socket lives under it — a unix socket path is 103 characters, and the + // connector refuses a longer one. + dir, err := os.MkdirTemp("", "bcrh") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + require.NoError(t, os.Chmod(dir, 0o700)) + require.NoError(t, os.Mkdir(filepath.Join(dir, "sessions"), 0o700)) + require.NoError(t, os.Mkdir(filepath.Join(dir, "work"), 0o700)) + require.NoError(t, os.Mkdir(filepath.Join(dir, "work-other"), 0o700)) + sc.Driver = d.Name + h := &harness{t: t, dir: dir, state: harnessStateDir(t, dir), driver: d, sc: sc} + h.writeScenario() + + exe, exeErr := os.Executable() + require.NoError(t, exeErr) + h.agent = filepath.Join(dir, "agent") + wrapper := "#!/bin/sh\n" + + harnessAgentEnv + "=" + shellQuote(d.Name) + " " + harnessDirEnv + "=" + shellQuote(dir) + " exec " + shellQuote(exe) + ` "$@"` + "\n" + require.NoError(t, os.WriteFile(h.agent, []byte(wrapper), 0o700)) //nolint:gosec // the fake agent's wrapper must be executable + for _, name := range []string{feedFile, storeFile, linesFile, pollsFile, agentLogFile, liveFile, workspaceFile} { + require.NoError(t, os.WriteFile(filepath.Join(dir, name), nil, 0o600)) + } + t.Cleanup(h.killAgents) + return h +} + +// harnessStateDir is the connector's state directory for this harness: +// /state/basecamp/connect/-, which is what +// connector.StateRoot resolves to with XDG_STATE_HOME set to /state. +// The location and the name are both part of what a worker's MCP server +// checks, so the harness's directory is the real shape, not a temp name. +func harnessStateDir(t *testing.T, dir string) string { + t.Helper() + state := filepath.Join(dir, "state", "basecamp", "connect", StateDirName(harnessAccount, harnessAgent)) + require.NoError(t, os.MkdirAll(state, 0o700)) + for d := state; d != dir; d = filepath.Dir(d) { + require.NoError(t, os.Chmod(d, 0o700)) + } + return state +} + +func shellQuote(s string) string { return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" } + +func (h *harness) writeScenario() { + data, err := json.Marshal(h.sc) + require.NoError(h.t, err) + require.NoError(h.t, os.WriteFile(filepath.Join(h.dir, scenarioFile), data, 0o600)) +} + +// Files in a harness directory. +const ( + scenarioFile = "scenario.json" + feedFile = "feed.jsonl" + storeFile = "basecamp.jsonl" + linesFile = "lines.jsonl" + pollsFile = "polls.jsonl" + agentLogFile = "agent.jsonl" + liveFile = "live.jsonl" + workspaceFile = "workspaces.jsonl" + // tokensDir holds the task tokens the fake workers were handed, so the + // parent can look for them everywhere a token must not be. + tokensDir = "tokens" + // connectorFile is the running connector's own identity: the pid a fake + // worker kills, so no process this harness did not start is signaled. + connectorFile = "connector.json" +) + +func readScenario(dir string) (harnessScenario, error) { + var sc harnessScenario + data, err := os.ReadFile(filepath.Join(dir, scenarioFile)) + if err != nil { + return sc, err + } + return sc, json.Unmarshal(data, &sc) +} + +// harnessRun is one start of the connector. +type harnessRun struct { + // Kill names where the connector kills itself; empty runs it to Until. + Kill string + // Until is the ledger predicate a surviving run stops at: "settled" + // when empty. + Until string + // SpawnFail is how many of this run's worker starts fail before any + // process exists. + SpawnFail int + // Filters is the feed filter set, as JSON; none when empty. + Filters string + // Killed says the run is expected to die by SIGKILL, from the connector + // itself or from the fake agent. + Killed bool + // StateDir is the connector's state directory; the harness's own + // (harness.state) when empty. + StateDir string + // Fault is a standing misbehavior of the fake Basecamp for the run: + // "stall-catch-up" holds the feed's first poll until the socket has + // served every live event; "repair-stall" never answers a repair poll. + Fault string + // Env is added to the connector's environment. + Env []string + // NoDispatch runs intake, admission and hooks but no dispatcher or + // outbox: a connector that dies after admitting, before its dispatcher + // could have seen the record, without racing one that might. + NoDispatch bool + // NoOutbox runs the dispatcher without the outbox: a connector that dies + // after settling an attempt, before anything could have claimed its + // notice. + NoOutbox bool + // RequireLog is a line the connector must have written by the end of the + // run: what it decided, where the ledger cannot show that it decided + // anything (a held attempt is indistinguishable from one recovery never + // looked at). + RequireLog string + // Shadow runs intake and admission only, and installs no hooks: a + // `--shadow` run. + Shadow bool +} + +// run starts the connector and waits for it to end as expected. +func (h *harness) run(r harnessRun) { + h.t.Helper() + cmd, out := h.start(r) + h.wait(cmd, out, r) +} + +// defaults fills a run in the same way for whoever starts it and whoever +// waits for it. +func (h *harness) defaults(r harnessRun) harnessRun { + if r.Killed && r.Until == "" { + // A run that is to die runs until it does. + r.Until = "never" + } + if r.StateDir == "" { + r.StateDir = h.state + } + return r +} + +func (h *harness) start(r harnessRun) (*exec.Cmd, *lockedBuffer) { + h.t.Helper() + r = h.defaults(r) + // Longer than any run's own deadline (runHarnessConnector's runFor), so + // a connector that overruns fails saying the ledger never got there + // rather than being killed by this timeout — which wait would otherwise + // be unable to tell from the kill a row asked for. + ctx, cancel := context.WithTimeout(context.Background(), harnessRunCap) + h.t.Cleanup(cancel) + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestRecoveryConnector$", "-test.count=1", "-test.v") + cmd.Env = append(os.Environ(), + harnessConnectorEnv+"=1", + harnessDirEnv+"="+h.dir, + harnessKillEnv+"="+r.Kill, + harnessUntilEnv+"="+r.Until, + harnessSpawnFailEnv+"="+strconv.Itoa(r.SpawnFail), + harnessFiltersEnv+"="+r.Filters, + harnessStateEnv+"="+r.StateDir, + "XDG_STATE_HOME="+filepath.Join(h.dir, "state"), + harnessShadowEnv+"="+strconv.FormatBool(r.Shadow), + harnessFaultEnv+"="+r.Fault, + harnessNoDispatchEnv+"="+strconv.FormatBool(r.NoDispatch), + harnessNoOutboxEnv+"="+strconv.FormatBool(r.NoOutbox), + ) + cmd.Env = append(cmd.Env, r.Env...) + out := &lockedBuffer{} + cmd.Stdout, cmd.Stderr = out, out + require.NoError(h.t, cmd.Start()) + h.watchForTokenFiles() + return cmd, out +} + +func (h *harness) wait(cmd *exec.Cmd, out *lockedBuffer, r harnessRun) { + h.t.Helper() + r = h.defaults(r) + started := time.Now() + err := cmd.Wait() + // exec.CommandContext kills with SIGKILL as well, and wait must not read + // that as the kill a row asked for. + require.Less(h.t, time.Since(started), harnessRunCap, "the connector outran the harness's own deadline\n%s", out.String()) + defer h.requireNoTaskTokenLeaked(out, r.StateDir) + if path := os.Getenv("BASECAMP_RECOVERY_DEBUG"); path != "" { + // Appended: a test is several runs, and the one that matters is + // rarely the last. + if f, openErr := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600); openErr == nil { + _, _ = f.WriteString(out.String()) + _ = f.Close() + } + } + if r.Killed { + var exit *exec.ExitError + require.True(h.t, errors.As(err, &exit), "the connector must die (kill %q): %v\n%s", r.Kill, err, out.String()) + status, ok := exit.Sys().(syscall.WaitStatus) + require.True(h.t, ok) + require.True(h.t, status.Signaled() && status.Signal() == syscall.SIGKILL, "killed at %q, got %v\n%s", r.Kill, err, out.String()) + return + } + require.NoError(h.t, err, "the connector must run to %q and stop cleanly\n%s", r.Until, out.String()) + if r.RequireLog != "" { + require.Contains(h.t, out.String(), r.RequireLog, "the connector said what it decided") + } +} + +type lockedBuffer struct { + mu sync.Mutex + buf []byte +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.buf = append(b.buf, p...) + return len(p), nil +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return string(b.buf) +} + +// killAgents ends every fake agent a harness started that is still alive, so +// a failed test leaves nothing behind. It signals by the identity the agent +// recorded — pid, group and start time — through the same call the connector +// uses, so the harness never signals a pid the kernel has since reused. +func (h *harness) killAgents() { + for _, entry := range h.agentLog() { + if entry.Step != "start" || entry.PID <= 0 || entry.PGID <= 0 { + continue + } + _, _ = driver.TerminateRecorded(driver.Process{PID: entry.PID, PGID: entry.PGID, StartedAt: entry.StartedAt}, time.Second) + } + // A worker's own children, which a group signal cannot reach once the + // worker that led the group is gone. + for _, child := range h.children() { + if owns, err := driver.OwnsWorker(driver.Process{PID: child.PID, PGID: child.PID, StartedAt: child.StartedAt}); err == nil && owns { + _ = syscall.Kill(child.PID, syscall.SIGKILL) + } + } +} + +// watchForTokenFiles watches, for the rest of this run, every place a task +// token must never be written, for every token a worker takes while it runs. +// The workers watch too, but a worker the connector ends never reports; this +// watcher is the parent's, and always does. +func (h *harness) watchForTokenFiles() { + h.t.Helper() + h.watchMu.Lock() + defer h.watchMu.Unlock() + if h.watching == nil { + h.watching = map[string]func() []string{} + } + stop := make(chan struct{}) + h.watchStop = stop + go func() { + for { + select { + case <-stop: + return + case <-time.After(5 * time.Millisecond): + } + for _, token := range h.knownTokens() { + h.watchMu.Lock() + if _, ok := h.watching[token]; !ok { + h.watching[token] = drivertest.WatchForSecretFiles(token, + h.workDir(), filepath.Join(h.dir, "work-other"), filepath.Join(h.dir, "sessions")) + } + h.watchMu.Unlock() + } + } + }() +} + +// knownTokens reads the tokens the workers have taken so far, ignoring a +// directory that does not exist yet. +func (h *harness) knownTokens() []string { + entries, err := os.ReadDir(filepath.Join(h.dir, tokensDir)) + if err != nil { + return nil + } + var out []string + for _, e := range entries { + if token, err := os.ReadFile(filepath.Join(h.dir, tokensDir, e.Name())); err == nil && len(token) > 0 { + out = append(out, string(token)) + } + } + return out +} + +// stopWatchingForTokenFiles ends the watchers and reports what they saw. +func (h *harness) stopWatchingForTokenFiles() int { + h.watchMu.Lock() + defer h.watchMu.Unlock() + if h.watchStop != nil { + close(h.watchStop) + h.watchStop = nil + } + watched := len(h.watching) + for token, stop := range h.watching { + for _, found := range stop() { + h.t.Errorf("a task token was written to %s while the connector ran", found) + } + delete(h.watching, token) + } + return watched +} + +// requireNoTaskTokenLeaked holds every run to the credential rule, for every +// task token any worker was handed so far: not in an agent's argv or +// environment, not in anything the connector wrote (its stdout lines, its +// log, the lifecycle messages it posted, the polls it made, the workspace +// records), not in any file under a working directory or the state directory +// — and no worker saw one appear in those files while it ran. +func (h *harness) requireNoTaskTokenLeaked(out *lockedBuffer, stateDir string) { + t := h.t + t.Helper() + watched := h.stopWatchingForTokenFiles() + tokens := h.taskTokens() + log := h.agentLog() + // A worker that bound to its task took a token, and the harness kept it. + // If it did not, this check has nothing to look for, and says so rather + // than passing. + bound, unbound := 0, 0 + var places drivertest.Places + for _, e := range log { + places.Env = append(places.Env, e.Env...) + places.Args = append(places.Args, e.Args...) + switch { + case e.Step == "bound": + bound++ + case strings.HasPrefix(e.Step, "bind-failed:"): + unbound++ + case strings.HasPrefix(e.Step, "secret-file:"): + t.Errorf("a worker saw a task token written to %s", strings.TrimPrefix(e.Step, "secret-file:")) + case strings.HasPrefix(e.Step, "secret-declared:"): + t.Errorf("a worker's MCP server declaration carried the task token in %s", strings.TrimPrefix(e.Step, "secret-declared:")) + } + } + // Every task the connector launched minted a token and is checked here, + // whoever its worker was — a fake one, or a real agent through the + // bridge, which leaves no agent log at all. + require.Len(t, tokens, h.tasksLaunched(stateDir), "every task the connector launched left its token for this check") + require.Equal(t, h.workersStarted(), bound+unbound, + "every worker that started either took its task's token or said why it could not") + for _, token := range h.takenTokens() { + require.Contains(t, tokens, token, "a worker took a token the connector did not mint for its task") + } + if len(tokens) == 0 { + require.Zero(t, watched, "nothing was watched, because no task was launched") + // Nothing to check is a fact about the run, not a pass: a run with + // no worker (a kill before the spawn, a start that ran nothing) is + // the only way here. + require.Equal(t, h.workersStarted(), unbound, + "a worker that started either took a token or said why it could not") + return + } + places.Texts = append(places.Texts, out.String()) + for _, name := range []string{linesFile, storeFile, pollsFile, workspaceFile, agentLogFile} { + data, err := os.ReadFile(filepath.Join(h.dir, name)) + require.NoError(t, err) + places.Texts = append(places.Texts, string(data)) + } + files := 0 + for _, token := range tokens { + // Env, argv and everything the connector wrote, in this process. + drivertest.RequireNoSecret(t, token, places) + // Every file under the working, session and state directories, read + // by a process of its own, which reports what it could not read. The + // state directory holds a ledger this test may have open. + found, read := scanForSecret(t, token, h.workDir(), filepath.Join(h.dir, "work-other"), + filepath.Join(h.dir, "sessions"), filepath.Join(h.dir, "state")) + for _, path := range found { + t.Errorf("a task token is in a file: %s", path) + } + require.Positive(t, read, "the scan read files; a scan that read nothing has cleared nothing") + files += read + } + t.Logf("credential check: %d task tokens, %d files read, %d watched while the run went on", len(tokens), files, watched) +} + +// taskTokens is every token the connector minted for a task, as its launch +// hook recorded it. +func (h *harness) taskTokens() []string { return h.tokenFiles("*.token", "taken-") } + +// takenTokens is every token a fake worker took over the socket. +func (h *harness) takenTokens() []string { return h.tokenFiles("taken-*.token", "") } + +func (h *harness) tokenFiles(pattern, exclude string) []string { + h.t.Helper() + paths, err := filepath.Glob(filepath.Join(h.dir, tokensDir, pattern)) + require.NoError(h.t, err) + var out []string + for _, path := range paths { + if exclude != "" && strings.HasPrefix(filepath.Base(path), exclude) { + continue + } + token, err := os.ReadFile(path) + require.NoError(h.t, err) + require.NotEmpty(h.t, token) + out = append(out, string(token)) + } + return out +} + +// tasksLaunched is how many tasks the ledger in dir says were launched, each +// with a token of its own. The ledger is read as it is: a run that made none +// (a shadow, a crash before the first launch) leaves none to open, and this +// must not be what creates one. +func (h *harness) tasksLaunched(dir string) int { + h.t.Helper() + ctx := context.Background() + l, err := OpenLedgerReadOnly(ctx, filepath.Join(dir, LedgerFile)) + if errors.Is(err, os.ErrNotExist) { + return 0 + } + require.NoError(h.t, err) + defer func() { _ = l.Close() }() + var n int + require.NoError(h.t, l.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM tasks`).Scan(&n)) + return n +} + +// workersStarted counts the worker processes that reached their agent, which +// is every worker that could have been handed a token. +func (h *harness) workersStarted() int { + n := 0 + for _, e := range h.agentLog() { + if e.Step == "start" { + n++ + } + } + return n +} + +// scanForSecret reads every file under dirs, in a process of its own, and +// reports what it found and how much it read. A scan that could not read +// something says so, and the caller fails on it: a check that skips is not a +// check that passed. +// +// A process of its own because reading a SQLite database's files by another +// descriptor in a process that holds the database open drops SQLite's POSIX +// advisory locks on them; another process closing the database then resets +// the WAL under the held handle, which reads stale or fails. The secret goes +// over stdin, never argv. +func scanForSecret(t *testing.T, secret string, dirs ...string) (found []string, read int) { + t.Helper() + cmd := exec.CommandContext(context.Background(), os.Args[0], dirs...) + cmd.Env = append(os.Environ(), harnessScanEnv+"=1") + cmd.Stdin = strings.NewReader(secret) + out, err := cmd.Output() + require.NoError(t, err, "the secret scan ran") + read = -1 + for line := range strings.SplitSeq(strings.TrimSpace(string(out)), "\n") { + kind, rest, ok := strings.Cut(line, "\t") + if !ok { + continue + } + switch kind { + case "found": + found = append(found, rest) + case "unreadable": + t.Errorf("the token scan could not read %s, so it cleared nothing there", rest) + case "read": + n, convErr := strconv.Atoi(rest) + require.NoError(t, convErr) + read = n + } + } + require.GreaterOrEqual(t, read, 0, "the scan reported what it read") + return found, read +} + +// runSecretScan is the scanning process: the secret on stdin, the directories +// as arguments, a path per line for every file that contains the secret. +func runSecretScan(dirs []string) int { + secret, err := io.ReadAll(os.Stdin) + if err != nil || len(secret) == 0 { + return 2 + } + read := 0 + for _, dir := range dirs { + if err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + switch { + case errors.Is(err, fs.ErrNotExist): + // It was there when the directory was read and gone when the + // walk reached it. The parent's watcher is what covers a file + // that only exists for a moment. + return nil + case err != nil: + // A place the scan could not look is not a place it has + // cleared: the caller is told, and fails. The walk goes on, + // so one unreadable entry does not hide the rest. + fmt.Println("unreadable\t" + path + ": " + err.Error()) + return nil //nolint:nilerr // reported to the caller, which fails on it + case d.Type().IsRegular(): + data, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return nil + } else if err != nil { + fmt.Println("unreadable\t" + path + ": " + err.Error()) + return nil //nolint:nilerr // reported to the caller, which fails on it + } + read++ + if bytes.Contains(data, secret) { + fmt.Println("found\t" + path) + } + } + return nil + }); err != nil { + fmt.Println("unreadable\t" + dir + ": " + err.Error()) + } + } + fmt.Println("read\t" + strconv.Itoa(read)) + return 0 +} + +// workspaces is every preparation and release of a task's working directory. +func (h *harness) workspaces() []workspaceEvent { + h.t.Helper() + var out []workspaceEvent + require.NoError(h.t, readJSONLines(filepath.Join(h.dir, workspaceFile), func(line []byte) error { + var e workspaceEvent + if err := json.Unmarshal(line, &e); err != nil { + return err + } + out = append(out, e) + return nil + })) + return out +} + +// releasedDir says a task's working directory was handed back, which the +// one-owner rule allows only once its worker's process group is gone. +func (h *harness) releasedDir(dir string) bool { + for _, e := range h.workspaces() { + if e.Step == "finish" && e.WorkDir == dir { + return true + } + } + return false +} + +// workDir is the first routed project's working directory. +func (h *harness) workDir() string { return filepath.Join(h.dir, "work") } + +// children is every process a fake worker started of its own, with the time +// it started, so it can be signaled only while it is still that process. +func (h *harness) children() []driver.Process { + var out []driver.Process + for _, e := range h.agentLog() { + if e.Step == "grandchild" && e.Child > 0 { + out = append(out, driver.Process{PID: e.Child, PGID: e.PGID, StartedAt: e.StartedAt}) + } + } + return out +} + +// ledger opens the harness's ledger. The connector need not be stopped. +func (h *harness) ledger() *Ledger { + h.t.Helper() + l, err := OpenLedger(filepath.Join(h.state, LedgerFile)) + require.NoError(h.t, err) + h.t.Cleanup(func() { _ = l.Close() }) + return l +} diff --git a/internal/connector/recovery_hold_test.go b/internal/connector/recovery_hold_test.go new file mode 100644 index 000000000..14cd6c62b --- /dev/null +++ b/internal/connector/recovery_hold_test.go @@ -0,0 +1,240 @@ +//go:build unix + +package connector + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The hold and the cutover: a record still being read when the hold is set, +// or when a shadow is promoted, becomes held rather than dispatched, a crash +// anywhere in shadow promote or import leaves the untouched shadow or a held +// ledger, and no restart dispatches a held record. + +// awaitHarnessFile waits for a file a connector process writes. +func (h *harness) awaitFile(name string) { + h.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + require.NoError(h.t, awaitFile(ctx, filepath.Join(h.dir, name)), "waiting for %s", name) +} + +func (h *harness) release(name string) { + h.t.Helper() + require.NoError(h.t, os.WriteFile(filepath.Join(h.dir, name), nil, 0o600)) +} + +// assertNothingDispatched checks no attempt was ever made and no worker ever +// ran, and that each record is held. +func (h *harness) assertNothingDispatched(l *Ledger, held ...int64) { + t := h.t + t.Helper() + assert.Empty(t, harnessAttempts(t, l), "no attempt") + assert.Zero(t, h.agentStarts(), "no worker process") + for _, id := range held { + assert.Equal(t, StateHeld, stateOf(t, l, id), "event %d", id) + } + assert.Empty(t, h.connectorPosts(), "nothing posted under the hold") +} + +func TestRecoveryARecordReadDuringTheHoldIsHeld(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + for _, crash := range []bool{false, true} { + name := "the read completes" + if crash { + name = "the connector is killed mid-read and restarted" + } + t.Run(name, func(t *testing.T) { + h := newHarness(t, d, harnessScenario{ReadGate: []int64{5001}}) + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + run := harnessRun{Until: "state:101=held"} + if crash { + run = harnessRun{Until: "never", Killed: true} + } + cmd, out := h.start(run) + h.awaitFile("read-waiting-5001") + + l := h.ledger() + _, err := l.SetHold(context.Background(), "operator", HoldByOperator) + require.NoError(t, err) + assert.Equal(t, StateSeen, stateOf(t, l, 101), "the read is still in flight at the hold") + if crash { + require.NoError(t, cmd.Process.Kill()) + } else { + h.release("read-release-5001") + } + h.wait(cmd, out, run) + if crash { + h.sc.ReadGate = nil + h.writeScenario() + } + + // A supervisor restart, however many times. + h.run(harnessRun{}) + h.run(harnessRun{}) + h.assertNothingDispatched(l, 101) + + // Released, a held record stays held; a new one runs. + _, err = l.Release(context.Background(), "operator") + require.NoError(t, err) + h.publish(feedEntry{Event: todoEvent(102, 5002)}) + h.run(harnessRun{}) + assert.Equal(t, StateHeld, stateOf(t, l, 101)) + assert.Equal(t, 0, h.handed(101)) + assert.Equal(t, 1, h.handed(102), "the connector does dispatch what the hold does not hold") + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, 102)) + }) + } + }) +} + +// cutover is a shadow run's state directory and the normal one beside it. +type cutover struct { + shadowDir, stateDir string +} + +// newCutover lays the two directories out under the harness's own state home, +// where the connector puts them, so a worker's MCP server would resolve the +// promoted one exactly as it resolves an ordinary run's. +func newCutover(h *harness) cutover { + t := h.t + t.Helper() + root := filepath.Join(h.dir, "state", "basecamp") + c := cutover{ + shadowDir: filepath.Join(root, "connect-shadow", StateDirName(harnessAccount, harnessAgent)), + stateDir: filepath.Join(root, "connect", StateDirName(harnessAccount, harnessAgent)), + } + for _, dir := range []string{root, filepath.Dir(c.shadowDir), filepath.Dir(c.stateDir), c.shadowDir, c.stateDir} { + require.NoError(t, os.MkdirAll(dir, 0o700)) + require.NoError(t, os.Chmod(dir, 0o700)) + } + return c +} + +// shadowLedger runs a shadow connector that admits event 101 and is stopped +// while event 102's read is in flight. +func (h *harness) shadowLedger(c cutover) { + h.t.Helper() + h.sc.ReadGate = []int64{5002} + h.writeScenario() + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{Shadow: true, StateDir: c.shadowDir, Until: "state:101=admitted"}) + + h.publish(feedEntry{Event: todoEvent(102, 5002)}) + run := harnessRun{Shadow: true, StateDir: c.shadowDir, Until: "never", Killed: true} + cmd, out := h.start(run) + h.awaitFile("read-waiting-5002") + require.NoError(h.t, cmd.Process.Kill()) + h.wait(cmd, out, run) + + h.sc.ReadGate = nil + h.writeScenario() +} + +func (c cutover) normalLedgerExists() bool { + _, err := os.Lstat(filepath.Join(c.stateDir, LedgerFile)) + return err == nil +} + +func TestRecoveryACrashInShadowPromoteNeverDispatchesAHeldRecord(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + for _, step := range []string{"locked", "marker", "tagged", "held", "checkpointed", "renamed", "synced"} { + t.Run(step, func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + c := newCutover(h) + h.shadowLedger(c) + + runKilled(t, "promote:"+step, "SHADOW_DIR="+c.shadowDir, "STATE_DIR="+c.stateDir) + if c.normalLedgerExists() { + // The supervisor restarts the connector over whatever the + // crash left at the normal path. + t.Logf("killed at %q: the ledger is at the normal path, and a restart must dispatch nothing", step) + h.run(harnessRun{StateDir: c.stateDir}) + h.assertNothingDispatched(h.ledgerAt(c.stateDir), 101, 102) + } else { + t.Logf("killed at %q: the shadow is untouched or held, and there is nothing at the normal path to restart over", step) + assertUntouchedOrHeldShadow(t, c.shadowDir) + } + + got, err := PromoteShadow(context.Background(), PromoteOptions{ + ShadowDir: c.shadowDir, StateDir: c.stateDir, AccountID: harnessAccount, AgentID: harnessAgent, By: "operator", + }) + require.NoError(t, err) + assert.Equal(t, HoldByPromote, got.Hold.Cause) + h.run(harnessRun{StateDir: c.stateDir}) + h.run(harnessRun{StateDir: c.stateDir}) + l := h.ledgerAt(c.stateDir) + h.assertNothingDispatched(l, 101, 102) + }) + } + }) +} + +func TestRecoveryACrashInImportNeverDispatchesAHeldRecord(t *testing.T) { + forEachDriver(t, func(t *testing.T, d harnessDriver) { + raceSubset(t, false) + for _, step := range []string{"entry", "tagged"} { + t.Run(step, func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + c := newCutover(h) + h.shadowLedger(c) + _, err := PromoteShadow(context.Background(), PromoteOptions{ + ShadowDir: c.shadowDir, StateDir: c.stateDir, AccountID: harnessAccount, AgentID: harnessAgent, By: "operator", + }) + require.NoError(t, err) + + file := `{"version":1,"entries":[{"event_id":102,"decision":"done"},{"event_id":101,"decision":"held"}]}` + runKilled(t, "import:"+step, "LEDGER="+filepath.Join(c.stateDir, LedgerFile), "RECONCILIATION="+file) + h.run(harnessRun{StateDir: c.stateDir}) + l := h.ledgerAt(c.stateDir) + h.assertNothingDispatched(l, 101, 102) + + // The import run again applies whole, and still nothing runs. + r, err := ParseReconciliation([]byte(file)) + require.NoError(t, err) + _, err = l.Import(context.Background(), r, "operator") + require.NoError(t, err) + h.run(harnessRun{StateDir: c.stateDir}) + assert.Empty(t, harnessAttempts(t, l)) + assert.Equal(t, StateHeld, stateOf(t, l, 101)) + assert.Equal(t, StateDiscarded, stateOf(t, l, 102)) + }) + } + }) +} + +// assertUntouchedOrHeldShadow is the other half of the promote rule: what the +// crash left is a shadow ledger, held or exactly as it was — never an unheld +// ledger at the normal path, which the caller has already established is not +// there. +func assertUntouchedOrHeldShadow(t *testing.T, shadowDir string) { + t.Helper() + l, err := OpenLedgerReadOnly(context.Background(), filepath.Join(shadowDir, LedgerFile)) + require.NoError(t, err, "the shadow ledger is still where it was") + defer func() { _ = l.Close() }() + held, err := l.Held(context.Background()) + require.NoError(t, err) + state := stateOf(t, l, 101) + if held { + assert.Equal(t, StateHeld, state, "a held shadow holds its waiting record") + } else { + assert.Equal(t, StateAdmitted, state, "an untouched shadow is as the crash found it") + } +} + +func (h *harness) ledgerAt(dir string) *Ledger { + h.t.Helper() + l, err := OpenLedger(filepath.Join(dir, LedgerFile)) + require.NoError(h.t, err) + h.t.Cleanup(func() { _ = l.Close() }) + return l +} diff --git a/internal/connector/recovery_intake_test.go b/internal/connector/recovery_intake_test.go new file mode 100644 index 000000000..2d962e5a0 --- /dev/null +++ b/internal/connector/recovery_intake_test.go @@ -0,0 +1,345 @@ +//go:build unix + +package connector + +import ( + "context" + "encoding/json" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" +) + +// Intake's recovery, with the connector killed around the feed: a filter +// change, a saturated backlog, a live buffer overflow and a stalled repair +// walk. None of it depends on the driver, so these run once, with the first +// registered one. + +func intakeHarness(t *testing.T, sc harnessScenario) *harness { + t.Helper() + if testing.Short() { + t.Skip("starts processes") + } + raceSubset(t, false) + // Always the same row, whatever else registers: intake does not depend + // on the driver, and file order must not pick one silently. + d, ok := harnessDriverNamed("claude") + require.True(t, ok, "the intake tests run with the claude row") + return newHarness(t, d, sc) +} + +// strangerEvent is an event by someone the agent does not trust: intake records +// it and admission discards it without a read, so a test can push thousands. +func strangerEvent(id int64) eventfeed.Event { + e := todoEvent(id, 9000+id%1000) + e.CreatorID = 4242 + return e +} + +func strangers(from, to int64, tweak func(*feedEntry)) []feedEntry { + out := make([]feedEntry, 0, to-from+1) + for id := from; id <= to; id++ { + e := feedEntry{Event: strangerEvent(id)} + if tweak != nil { + tweak(&e) + } + out = append(out, e) + } + return out +} + +func (h *harness) feedKey(filters eventfeed.Filters) eventfeed.CheckpointKey { + h.t.Helper() + origin, err := eventfeed.CanonicalOrigin(harnessOrigin) + require.NoError(h.t, err) + return eventfeed.CheckpointKey{Origin: origin, AccountID: harnessAccount, ConsumerNamespace: harnessNamespace, FilterKey: filters.FilterKey()} +} + +// positionID is the id a stored feed position is after; zero for none. +func (h *harness) positionID(l *Ledger, filters eventfeed.Filters) int64 { + h.t.Helper() + position, ok, err := l.Load(context.Background(), h.feedKey(filters)) + require.NoError(h.t, err) + if !ok { + return 0 + } + require.True(h.t, strings.HasPrefix(position, "feed-"), "the feed's checkpoint is a feed position, never a repair walk's: %q", position) + id, err := strconv.ParseInt(strings.TrimPrefix(position, "feed-"), 10, 64) + require.NoError(h.t, err) + return id +} + +func (h *harness) feedPolls() []pollLog { + var out []pollLog + for _, p := range h.polls() { + if !p.Repair { + out = append(out, p) + } + } + return out +} + +// A filter change re-enters after the last poll-served id, not at the present, +// and a crash before the new filter set saves a position re-enters there again. +func TestRecoveryAFilterChangeResumesFromTheLastPollServedID(t *testing.T) { + h := intakeHarness(t, harnessScenario{}) + h.publish(strangers(101, 103, nil)...) + h.run(harnessRun{}) + + h.publish(strangers(104, 105, nil)...) + narrowed := eventfeed.Filters{Buckets: []int64{harnessBucket}} + raw, err := json.Marshal(narrowed) + require.NoError(t, err) + before := len(h.feedPolls()) + h.run(harnessRun{Filters: string(raw), Kill: "feed-poll", Killed: true}) + l := h.ledger() + assert.Zero(t, h.positionID(l, narrowed), "killed before the new filter set polled") + + h.run(harnessRun{Filters: string(raw)}) + polls := h.feedPolls()[before:] + require.NotEmpty(t, polls) + assert.Equal(t, "103", polls[0].Since, "entered after the last id the poll lane served under the old filters") + assert.Empty(t, polls[0].Position) + for _, id := range []int64{104, 105} { + _, ok, err := l.Get(context.Background(), id) + require.NoError(t, err) + assert.True(t, ok, "event %d after the filter change is not skipped", id) + } + assert.Equal(t, int64(105), h.positionID(l, narrowed)) +} + +// A saturated backlog stops intake reading the feed without moving the +// checkpoint past what admission has not taken; a crash there loses nothing. +func TestRecoveryBacklogSaturationPausesTheFeedWithoutMovingTheCheckpoint(t *testing.T) { + var gate []int64 + var events []feedEntry + for id := int64(101); id <= 110; id++ { + recording := 5000 + id + gate = append(gate, recording) + events = append(events, feedEntry{Event: todoEvent(id, recording)}) + } + h := intakeHarness(t, harnessScenario{QueueWarn: 1, QueuePause: 2, ReadGate: gate}) + h.publish(events...) + h.run(harnessRun{Kill: "paused", Killed: true}) + + l := h.ledger() + assert.Zero(t, h.positionID(l, eventfeed.Filters{}), "the page behind the pause was never checkpointed") + served, err := l.LastPollServedID(context.Background(), h.feedKey(eventfeed.Filters{})) + require.NoError(t, err) + assert.Zero(t, served) + assert.Empty(t, harnessAttempts(t, l)) + + h.sc.ReadGate = nil + h.writeScenario() + h.run(harnessRun{}) + for _, e := range events { + assert.Equal(t, 1, h.handed(e.Event.ID), "event %d", e.Event.ID) + assert.Equal(t, string(OutcomeSucceeded), outcomeOf(t, l, e.Event.ID), "event %d", e.Event.ID) + } + assert.Equal(t, int64(110), h.positionID(l, eventfeed.Filters{})) +} + +// A live buffer overflow is on disk before it is accepted, the retained events +// are drained, and the repair walk recovers the dropped ids on its own cursor: +// the feed's checkpoint never moves to a live id, so the unpolled range behind +// it is still served; a crash between the signal and the walk resumes the walk +// on start; the walk repeats through the safety delay, and what it never +// serves is recorded unrecovered once the window closes. +func TestRecoveryABufferOverflowIsReconciledAcrossACrash(t *testing.T) { + // The buffer reports each drop as it happens: two drops, two losses. + h := intakeHarness(t, harnessScenario{RepairWindow: 1500 * time.Millisecond, OverflowLosses: 2}) + h.publish(strangers(101, 103, nil)...) + h.run(harnessRun{}) + + const ( + straggler = int64(60_001) // dropped; poll-visible from the third repair poll + deleted = int64(60_002) // dropped; never poll-visible + lastLive = int64(70_002) + ) + // Behind the live burst, an unpolled range the feed has not served yet. + behind := strangers(104, 106, func(e *feedEntry) { e.FromRepairPoll = 1 }) + // Ten thousand and two live events: the buffer holds ten thousand, and + // drops the two oldest. + burst := strangers(straggler, lastLive, func(e *feedEntry) { + e.Live, e.FromRepairPoll = true, 1 + switch e.Event.ID { + case straggler: + e.FromRepairPoll = 3 + case deleted: + e.Never = true + } + }) + h.publish(append(behind, burst...)...) + h.run(harnessRun{Fault: "stall-catch-up", Kill: "repair-poll", Killed: true}) + + l := h.ledger() + losses, err := l.OpenLosses(context.Background()) + require.NoError(t, err) + require.Len(t, losses, 2, "the overflow was written down before the walk began") + var missing []int64 + for _, loss := range losses { + ids, err := l.MissingIDs(context.Background(), loss.ID, LossMissing) + require.NoError(t, err) + require.Len(t, ids, 1) + assert.Equal(t, ids[0]-1, loss.RepairSince, "a walk enters just before its missing id") + missing = append(missing, ids...) + } + slices.Sort(missing) + assert.Equal(t, []int64{straggler, deleted}, missing) + assert.LessOrEqual(t, h.positionID(l, eventfeed.Filters{}), int64(103), "no live id moved the checkpoint") + + // Every checkpoint the ledger holds while the walk runs, not only the one + // it ends with: a walk that wrote its own cursor there would be overwritten + // by the feed's next page. + positions := h.watchCheckpoints() + h.run(harnessRun{Until: "losses-closed"}) + sampled := positions() + // The watcher must have been watching for the whole run, or it proves + // nothing: it saw the position the run started from and the one it ended + // at, both of which stand long enough to be seen, and positions between. + require.Contains(t, sampled, "feed-103", "the watcher saw the run start") + require.Contains(t, sampled, "feed-70002", "the watcher saw the run end") + require.Greater(t, len(sampled), 2, "the watcher saw the checkpoint move") + for _, position := range sampled { + assert.True(t, strings.HasPrefix(position, "feed-"), "the feed's checkpoint only ever holds a feed position, saw %q", position) + } + + for _, id := range []int64{104, 105, 106} { + r, ok, err := l.Get(context.Background(), id) + require.NoError(t, err) + require.True(t, ok, "event %d behind the live burst is still served", id) + assert.Equal(t, LanePoll, r.Lane, "event %d came from the feed's own walk", id) + } + _, ok, err := l.Get(context.Background(), straggler) + require.NoError(t, err) + require.True(t, ok, "the straggler was recovered") + unrecovered, err := l.UnrecoveredIDs(context.Background()) + require.NoError(t, err) + assert.Equal(t, []int64{deleted}, unrecovered) + // The checkpoint is the feed's own walk, wherever the repair walk got to. + assert.Equal(t, lastLive, h.positionID(l, eventfeed.Filters{})) + + var walks, servedAt int + for _, p := range h.polls() { + if !p.Repair { + assert.False(t, strings.HasPrefix(p.Position, "repair-"), "the feed never walks from a repair cursor") + continue + } + walks++ + if slices.Contains(p.Served, straggler) && servedAt == 0 { + servedAt = walks + } + } + // The straggler is withheld until the third repair poll; that its loss + // is recovered rather than unrecovered is the walk having repeated + // through the safety delay, and a walk still polling after serving it + // is the walk repeating until the window closed for the id that never + // came. + assert.Greater(t, walks, max(servedAt, 1), "the walk repeated, and kept repeating until the window closed") + var recovered []int64 + for _, loss := range losses { + ids, err := l.MissingIDs(context.Background(), loss.ID, LossRecovered) + require.NoError(t, err) + recovered = append(recovered, ids...) + } + assert.Equal(t, []int64{straggler}, recovered, "the straggler's loss is recovered, whichever walk served it first") + + // The unpolled range behind the burst is served before anything from the + // burst: a checkpoint taken from a live id would have skipped it. + // In the order the feed's own walk served events, not by page: one page + // may carry both. + behindAt, aheadAt, n := -1, -1, 0 + for _, p := range h.feedPolls() { + for _, id := range p.Served { + if id == 106 && behindAt < 0 { + behindAt = n + } + if id >= straggler && aheadAt < 0 { + aheadAt = n + } + n++ + } + } + require.GreaterOrEqual(t, behindAt, 0, "the feed's own walk served the range behind the burst") + require.GreaterOrEqual(t, aheadAt, 0, "and went on past it") + assert.Less(t, behindAt, aheadAt, "the feed never jumped a live id ahead of the range behind it") +} + +// watchCheckpoints samples the feed's stored position until the returned +// function is called, which returns everything it saw. +func (h *harness) watchCheckpoints() func() []string { + stop := make(chan struct{}) + done := make(chan []string, 1) + go func() { + seen := map[string]bool{} + for { + select { + case <-stop: + out := make([]string, 0, len(seen)) + for position := range seen { + out = append(out, position) + } + done <- out + return + case <-time.After(2 * time.Millisecond): + } + l, err := OpenLedgerReadOnly(context.Background(), filepath.Join(h.state, LedgerFile)) + if err != nil { + continue + } + rows, err := l.db.QueryContext(context.Background(), `SELECT position FROM checkpoints`) + if err == nil { + for rows.Next() { + var position string + if rows.Scan(&position) == nil { + seen[position] = true + } + } + _ = rows.Close() + } + _ = l.Close() + } + }() + return func() []string { + close(stop) + select { + case out := <-done: + return out + case <-time.After(10 * time.Second): + h.t.Fatal("the checkpoint watcher did not stop") + return nil + } + } +} + +// A repair walk that never answers holds up nothing: live events still reach +// the ledger while the loss stays open. +func TestRecoveryAStalledRepairWalkDoesNotStopLiveIntake(t *testing.T) { + h := intakeHarness(t, harnessScenario{RepairWindow: time.Hour}) + h.publish(strangers(101, 101, nil)...) + h.run(harnessRun{}) + + l := h.ledger() + _, err := l.RecordLoss(context.Background(), []int64{90_001}, time.Now(), time.Hour, eventfeed.Filters{}) + require.NoError(t, err) + h.publish(strangers(90_005, 90_005, func(e *feedEntry) { e.Live, e.Never = true, true })...) + h.publish(strangers(102, 102, nil)...) + h.run(harnessRun{Fault: "repair-stall", Until: "state:90005,state:102"}) + + losses, err := l.OpenLosses(context.Background()) + require.NoError(t, err) + assert.Len(t, losses, 1, "the loss is still open") + stalled := false + for _, p := range h.polls() { + stalled = stalled || p.Stalled + } + assert.True(t, stalled, "the repair walk was running, and stalled") +} diff --git a/internal/connector/recovery_norace_test.go b/internal/connector/recovery_norace_test.go new file mode 100644 index 000000000..24afde241 --- /dev/null +++ b/internal/connector/recovery_norace_test.go @@ -0,0 +1,5 @@ +//go:build unix && !race + +package connector + +const harnessUnderRace = false diff --git a/internal/connector/recovery_race_test.go b/internal/connector/recovery_race_test.go new file mode 100644 index 000000000..5a0ccc44f --- /dev/null +++ b/internal/connector/recovery_race_test.go @@ -0,0 +1,5 @@ +//go:build unix && race + +package connector + +const harnessUnderRace = true diff --git a/internal/connector/recovery_real_test.go b/internal/connector/recovery_real_test.go new file mode 100644 index 000000000..432eedd76 --- /dev/null +++ b/internal/connector/recovery_real_test.go @@ -0,0 +1,161 @@ +//go:build unix + +package connector + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/auth" +) + +// TestRecoveryAgainstRealAgents runs the kill points the connector itself can +// reach against the real agent binaries, with the real `basecamp mcp` built +// from this tree as the workers' MCP server. The server holds a token that +// reaches no Basecamp, so the workers' basecamp_connect calls are real and +// their Basecamp calls fail; nothing is posted anywhere. A model is called, so +// it is opt-in: +// +// make build +// BASECAMP_RECOVERY_REAL_AGENTS=1 BASECAMP_RECOVERY_BASECAMP=$PWD/bin/basecamp \ +// go test ./internal/connector/ -run TestRecoveryAgainstRealAgents -v +// +// What a model does with its turn is not scripted, so the assertions are the +// guarantees that hold whatever it does. +func TestRecoveryAgainstRealAgents(t *testing.T) { + if os.Getenv(harnessRealEnv) == "" { + t.Skip("opt in with " + harnessRealEnv + "=1 and " + harnessRealBasecampEnv + "=: it runs the real agents, which call their models") + } + basecampBinary := os.Getenv(harnessRealBasecampEnv) + require.NotEmpty(t, basecampBinary, harnessRealBasecampEnv+" names the basecamp binary built from this tree") + info, err := os.Stat(basecampBinary) + require.NoError(t, err, "the basecamp binary is where %s says", harnessRealBasecampEnv) + require.NotZero(t, info.Mode()&0o111, "%s is executable", basecampBinary) + rows := []struct { + name string + kill string + // lost says the attempt is lost: the kill left it live. + lost bool + // held says recovery cannot identify the worker, and holds the + // attempt rather than settling it. + held bool + }{ + {name: "no crash"}, + {name: "attempt launching", kill: "line:dispatch:launching", held: true}, + {name: "attempt running", kill: "line:dispatch:running", lost: true}, + {name: "after get_dispatch", kill: "get-dispatch", lost: true}, + } + ran := false + for _, d := range harnessDrivers { + if !d.Real { + continue + } + ran = true + t.Run(d.Name, func(t *testing.T) { + for _, row := range rows { + t.Run(row.name, func(t *testing.T) { + h := newHarness(t, d, harnessScenario{}) + stateDir := h.state + config := filepath.Join(h.dir, "config", "basecamp") + require.NoError(t, os.MkdirAll(config, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(config, "config.json"), + []byte(`{"profiles":{"agent":{"base_url":"http://127.0.0.1:9","account_id":"`+harnessAccount+`"}}}`), 0o600)) + // The worker's MCP server is the real one, and it refuses + // to start without a credential. The bridge hands it no + // token from the environment — by design — so the profile + // gets a stored one, in this harness's own config + // directory, pointing at a closed port. + t.Setenv("BASECAMP_NO_KEYRING", "1") + require.NoError(t, auth.NewStore(config).Save(auth.ProfileCredentialKey("agent"), &auth.Credentials{ + AccessToken: "test-token-not-real", OAuthType: "bc5", + ExpiresAt: time.Now().Add(time.Hour).Unix(), + })) + // These are appended after os.Environ(), and the last + // duplicate wins in exec, so what the operator's own + // environment says is overridden rather than reaching the + // worker's MCP server: a real BASECAMP_TOKEN by the fake + // one, and a BASECAMP_BASE_URL — which the server's + // environment allowlist passes, and which outranks the + // profile — by the same closed port the profile names. No + // request the server makes can leave the machine. + const closedPort = "http://127.0.0.1:9" + env := []string{ + harnessRealBasecampEnv + "=" + basecampBinary, + "XDG_CONFIG_HOME=" + filepath.Join(h.dir, "config"), + "BASECAMP_TOKEN=test-token-not-real", + "BASECAMP_BASE_URL=" + closedPort, + "BASECAMP_NO_KEYRING=1", + } + h.publish(feedEntry{Event: todoEvent(101, 5001)}) + h.run(harnessRun{StateDir: stateDir, Env: env, Kill: row.kill, Killed: row.kill != ""}) + + l := h.ledgerAt(stateDir) + var pids []int + for _, a := range harnessAttempts(t, l) { + var pid int + require.NoError(t, l.db.QueryRowContext(context.Background(), `SELECT COALESCE(pid, 0) FROM attempts WHERE id = ?`, a.ID).Scan(&pid)) + if pid > 0 { + pids = append(pids, pid) + } + } + if row.held { + // The other project's attempt is a second one; the + // held attempt is the first. + // Run until a real worker has finished an event in the + // other project: recovery returned and the dispatcher + // went on around the held attempt. + h.publish(feedEntry{Event: otherTodoEvent(102, 6001)}) + h.run(harnessRun{StateDir: stateDir, Env: env, Until: "state:102=completed"}) + attempts := harnessAttempts(t, l) + require.NotEmpty(t, attempts) + assert.Equal(t, string(AttemptLaunching), attempts[0].State, "held, not settled") + assert.Equal(t, StateDispatched, stateOf(t, l, 101)) + assert.Empty(t, h.notices(101)) + return + } + for range 2 { + h.run(harnessRun{StateDir: stateDir, Env: env}) + } + + attempts := harnessAttempts(t, l) + require.Len(t, attempts, 1, "no second attempt, whatever the worker did") + // A real agent ran, or this row proved nothing about one: + // the ledger recorded its process. + require.NotEmpty(t, pids, "the real agent's process was recorded") + assert.Equal(t, StateCompleted, stateOf(t, l, 101)) + outcome := outcomeOf(t, l, 101) + assert.True(t, slices.Contains([]string{string(OutcomeUnknown), string(OutcomeSucceeded), string(OutcomeFailed)}, outcome), "outcome %q", outcome) + if row.lost { + assert.Equal(t, string(StopLost), attempts[0].StopReason) + } + if row.kill == "line:dispatch:running" { + assert.Equal(t, string(OutcomeUnknown), outcome, "never prompted, still unknown: a process may have existed") + } + assert.LessOrEqual(t, len(h.notices(101)), 1, "at most one completion notice") + if row.kill == "" { + // The whole chain ran: the agent started its MCP + // server, the bridge took the task token from the + // connector's socket, and the worker called + // get_dispatch, which cancels the guard. + var canceled int + require.NoError(t, l.db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM task_events WHERE event_id = 101 AND guard = 'canceled'`).Scan(&canceled)) + assert.Equal(t, 1, canceled, "the real worker read its dispatch") + } + for _, pid := range pids { + assert.True(t, processGone(context.Background(), pid), "the worker the crash left is gone, pid %d", pid) + } + t.Logf("%s: outcome %s, stop %s, notices %d", row.name, outcome, attempts[0].StopReason, len(h.notices(101))) + }) + } + }) + } + require.True(t, ran, "no real agent registered") +} diff --git a/internal/connector/recovery_worker_test.go b/internal/connector/recovery_worker_test.go new file mode 100644 index 000000000..e0ccc48ff --- /dev/null +++ b/internal/connector/recovery_worker_test.go @@ -0,0 +1,474 @@ +//go:build unix + +package connector + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "regexp" + "slices" + "strconv" + "strings" + "syscall" + "time" + + "github.com/basecamp/basecamp-cli/internal/connector/driver" +) + +// The fake worker: what every fake agent does with a prompt, whatever its wire. + +// fakeWorker is what a worker does, whatever its wire: it reads its dispatch, +// acknowledges, replies and completes through the task token its MCP server +// declaration carries, as scripted per event. +type fakeWorker struct { + dir string + sc harnessScenario + ledger *Ledger + dispatch *TaskDispatch + + replies map[int64]int64 +} + +// agentLogEntry is one thing a fake agent did. +type agentLogEntry struct { + PID int `json:"pid"` + PGID int `json:"pgid"` + StartedAt time.Time `json:"started_at"` + Event int64 `json:"event,omitempty"` + N int `json:"n,omitempty"` + Step string `json:"step"` + // Args and Env are the agent's own, on its "start" entry. + Args []string `json:"args,omitempty"` + Env []string `json:"env,omitempty"` + // Child is the pid of the process a "grandchild" step started. + Child int `json:"child,omitempty"` + // Prompt is the prompt as the agent received it, on a "prompt" step. + Prompt string `json:"prompt,omitempty"` +} + +func newFakeWorker(dir string) (*fakeWorker, error) { + if dir == "" { + return nil, errors.New("no harness directory") + } + sc, err := readScenario(dir) + if err != nil { + return nil, err + } + w := &fakeWorker{dir: dir, sc: sc, replies: map[int64]int64{}} + pgid, _ := syscall.Getpgid(0) + // What this agent was started with, for the parent to check no task + // token is in either. + _ = appendJSONLine(filepath.Join(dir, agentLogFile), agentLogEntry{ + PID: os.Getpid(), PGID: pgid, StartedAt: time.Now(), Step: "start", Args: os.Args[1:], Env: os.Environ(), + }) + return w, nil +} + +func (w *fakeWorker) close() { + if w.ledger != nil { + _ = w.ledger.Close() + } +} + +func (w *fakeWorker) log(event int64, n int, step string) { + pgid, _ := syscall.Getpgid(0) + _ = appendJSONLine(filepath.Join(w.dir, agentLogFile), + agentLogEntry{PID: os.Getpid(), PGID: pgid, StartedAt: time.Now(), Event: event, N: n, Step: step}) +} + +func (h *harness) agentLog() []agentLogEntry { + var out []agentLogEntry + _ = readJSONLines(filepath.Join(h.dir, agentLogFile), func(line []byte) error { + var e agentLogEntry + if json.Unmarshal(line, &e) == nil { + out = append(out, e) + } + return nil + }) + return out +} + +// BadMode says this agent process reports a permission mode other than the +// one asked for. +func (w *fakeWorker) BadMode() bool { + starts := 0 + _ = readJSONLines(filepath.Join(w.dir, agentLogFile), func(line []byte) error { + var e agentLogEntry + if json.Unmarshal(line, &e) == nil && e.Step == "start" { + starts++ + } + return nil + }) + return starts <= w.sc.BadModeStarts +} + +// Bind takes the worker's task from the MCP server declaration its driver +// handed the agent, as the two commands behind that declaration do: +// +// - the declaration must name the bridge (`basecamp connect worker-mcp`) +// with the agent's profile, its state directory and its token socket, +// since the real bridge refuses without any of them +// (internal/commands/connect_worker_mcp.go); +// - the task token comes from that one-use socket, never from the +// environment; +// - the state directory is resolved by location and name, which is where +// the agent's id comes from, and the ledger is opened as it is, never +// created and never migrated — the connector owns it +// (internal/commands/mcp.go). +func (w *fakeWorker) Bind(ctx context.Context, server driver.MCPServer) error { + // Bound or not, the parent is told which: a worker that started and took + // no token must not look to the credential check like a run with nothing + // to check. + err := w.bind(ctx, server) + if err != nil { + w.log(0, 0, "bind-failed: "+err.Error()) + } + return err +} + +func (w *fakeWorker) bind(ctx context.Context, server driver.MCPServer) error { + if server.Name != MCPServerName { + return fmt.Errorf("the MCP server is %q, not %q", server.Name, MCPServerName) + } + if len(server.Args) < 2 || server.Args[0] != "connect" || server.Args[1] != "worker-mcp" { + return fmt.Errorf("the MCP server is not the connector's bridge: %v", server.Args) + } + if profile := flagValue(server.Args, "--profile"); profile == "" { + return errors.New("the MCP server names no profile, which the bridge refuses to start without") + } + stateArg := flagValue(server.Args, "--connect-state") + if stateArg == "" { + return errors.New("the MCP server has no --connect-state") + } + stateDir := stateArg + // The server resolves the directory against its own state home, which + // is the environment the driver declared for it, not this agent's: a + // declaration without one would send the real server elsewhere. + home, ok := server.Env["XDG_STATE_HOME"] + if !ok || home == "" { + return errors.New("the MCP server's environment declares no XDG_STATE_HOME") + } + if err := os.Setenv("XDG_STATE_HOME", home); err != nil { + return err + } + agentID, err := ResolveStateDir(stateDir, harnessAccount) + if err != nil { + return err + } + + token, err := w.takeToken(ctx, server.Args) + if err != nil { + return err + } + // The socket is the token's one carriage: the declaration that named the + // socket must not also carry the token. + for key, value := range server.Env { + if strings.Contains(value, token) { + w.log(0, 0, "secret-declared:env "+key) + } + } + for _, arg := range server.Args { + if strings.Contains(arg, token) { + w.log(0, 0, "secret-declared:argv") + } + } + l, err := OpenExistingLedger(ctx, filepath.Join(stateDir, LedgerFile)) + if err != nil { + return err + } + d, err := l.Dispatch(ctx, token, agentID) + if err != nil { + _ = l.Close() + return err + } + w.ledger, w.dispatch = l, d + if err := w.watchToken(token); err != nil { + return err + } + // Said only once the token is on disk for the parent's check and the + // watch is running: a worker that bound without either would leave the + // parent nothing to check. + w.log(0, 0, "bound") + return nil +} + +// flagValue is the value after name in a command line, empty when it is not +// there. +func flagValue(args []string, name string) string { + i := slices.Index(args, name) + if i < 0 || i+1 >= len(args) { + return "" + } + return args[i+1] +} + +// takeToken takes the task token from the connector's one-use socket, as the +// bridge the connector names does (`basecamp connect worker-mcp`, see +// internal/commands/connect_worker_mcp.go): the socket path is in the +// server's arguments, the token is a line on the socket, and it is served +// only to the worker's own process group — which this agent leads. +func (w *fakeWorker) takeToken(ctx context.Context, args []string) (string, error) { + socket := flagValue(args, "--socket") + if socket == "" { + return "", errors.New("the MCP server has no --socket") + } + dialer := net.Dialer{Timeout: 30 * time.Second} + conn, err := dialer.DialContext(ctx, "unix", socket) + if err != nil { + return "", fmt.Errorf("the connector's token socket: %w", err) + } + defer func() { _ = conn.Close() }() + _ = conn.SetDeadline(time.Now().Add(30 * time.Second)) + 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 +} + +// watchToken records that this worker took its token, and checks the one +// thing only the worker can: that the token it was handed is the one the +// connector minted for its attempt. +// +// The watch for a token in a file is the parent's, not this process's. A +// worker the connector ends by its process group takes any deferred report +// with it, and the crash rows end workers exactly that way; the parent is +// never killed, so it always reports. This process also holds the ledger +// open, and reading the ledger's own files here would drop SQLite's POSIX +// locks on them. +func (w *fakeWorker) watchToken(token string) error { + tokens := filepath.Join(w.dir, tokensDir) + if err := os.MkdirAll(tokens, 0o700); err != nil { + return err + } + f, err := os.CreateTemp(tokens, "taken-*.token") + if err != nil { + return err + } + if _, err := f.WriteString(token); err != nil { + _ = f.Close() + return err + } + return f.Close() +} + +var promptEvent = regexp.MustCompile(`Event (\d+)`) + +// Turn does what the scenario scripts for the prompt's event. An error is a +// step that could not be done; the agent reports the turn failed. +func (w *fakeWorker) Turn(ctx context.Context, prompt string) error { + m := promptEvent.FindStringSubmatch(prompt) + if m == nil { + return errors.New("the prompt names no event") + } + event, _ := strconv.ParseInt(m[1], 10, 64) + n := w.prompted(event) + 1 + pgid, _ := syscall.Getpgid(0) + _ = appendJSONLine(filepath.Join(w.dir, agentLogFile), + agentLogEntry{PID: os.Getpid(), PGID: pgid, StartedAt: time.Now(), Event: event, N: n, Step: "prompt", Prompt: prompt}) + steps, ok := w.sc.Plans[m[1]+"#"+strconv.Itoa(n)] + if !ok { + steps = []string{"get", "ack", "reply", "complete"} + } + for _, step := range steps { + if err := w.step(ctx, event, n, step); err != nil { + w.log(event, n, "error:"+step) + return fmt.Errorf("%s: %w", step, err) + } + w.log(event, n, step) + } + return nil +} + +func (w *fakeWorker) prompted(event int64) int { + n := 0 + _ = readJSONLines(filepath.Join(w.dir, agentLogFile), func(line []byte) error { + var e agentLogEntry + if json.Unmarshal(line, &e) == nil && e.Event == event && e.Step == "prompt" { + n++ + } + return nil + }) + return n +} + +func (w *fakeWorker) step(ctx context.Context, event int64, n int, step string) error { + if w.dispatch == nil { + return errors.New("the worker was never bound to a task") + } + name, arg, _ := strings.Cut(step, ":") + switch name { + case "get": + in, ok, err := w.dispatch.Get(ctx, event) + if err != nil { + return err + } + if !ok || in.EventID != event { + return fmt.Errorf("get_dispatch did not return event %d", event) + } + case "ack": + in, _, err := w.dispatch.Get(ctx, event) + if err != nil { + return err + } + id, err := postMessage(w.dir, storedMessage{Kind: MessageBoost, BucketID: in.Recording.BucketID, RecordingID: in.Recording.RecordingID, Content: "on it", By: "worker"}) + if err != nil { + return err + } + if _, err := w.dispatch.Ack(ctx, event, &id); err != nil { + return err + } + case "reply": + in, _, err := w.dispatch.Get(ctx, event) + if err != nil { + return err + } + id, err := postMessage(w.dir, storedMessage{Kind: MessageComment, BucketID: in.Recording.BucketID, RecordingID: in.ReplyTo.RecordingID, + Content: "done: event " + strconv.FormatInt(event, 10) + " attempt " + strconv.Itoa(n), By: "worker"}) + if err != nil { + return err + } + w.replies[event] = id + case "complete", "fail": + outcome := OutcomeSucceeded + if name == "fail" { + outcome = OutcomeFailed + } + c := Completion{Outcome: outcome} + if id, ok := w.replies[event]; ok { + c.ReplyID = &id + } + if _, err := w.dispatch.Complete(ctx, event, c); err != nil { + return err + } + case "kill": + // The connector dies while this worker is mid-turn. + return w.killConnector(ctx) + case "linger": + // A worker the connector left behind: it stays until something ends + // its process group, which only the connector's restart may do. + w.log(event, n, "linger") + time.Sleep(2 * time.Minute) + os.Exit(9) + case "exit": + code, _ := strconv.Atoi(arg) + os.Exit(code) + case "grandchild": + // A process of the worker's own that outlives it, in its process + // group, holding its working directory: the tree the one-owner rule + // says nothing may be released around. + wd, err := os.Getwd() + if err != nil { + return err + } + pid, err := syscall.ForkExec("/bin/sleep", []string{"sleep", "300"}, &syscall.ProcAttr{Dir: wd, Env: []string{}}) + if err != nil { + return err + } + pgid, _ := syscall.Getpgid(0) + return appendJSONLine(filepath.Join(w.dir, agentLogFile), + agentLogEntry{PID: os.Getpid(), PGID: pgid, StartedAt: time.Now(), Event: event, N: n, Step: "grandchild", Child: pid}) + case "arrive": + // A further event on the conversation while this one is in hand. + id, err := strconv.ParseInt(arg, 10, 64) + if err != nil { + return err + } + in, _, err := w.dispatch.Get(ctx, event) + if err != nil { + return err + } + return appendFeed(w.dir, feedEntry{Event: todoEvent(id, in.Recording.RecordingID)}) + case "await": + // Wait for a record to reach a state: "await:=". + id, state, _ := strings.Cut(arg, "=") + n, err := strconv.ParseInt(id, 10, 64) + if err != nil { + return err + } + return waitFor(ctx, func() (bool, error) { + r, ok, err := w.ledger.Get(ctx, n) + return ok && slices.Contains(strings.Split(state, "|"), string(r.State)), err + }) + default: + return fmt.Errorf("unknown step %q", step) + } + return nil +} + +// killConnector SIGKILLs the connector and returns once it is gone, so the +// steps after it run in a world without a connector. +// +// The connector is the pid it wrote down when it started, not this process's +// parent: a worker started behind an adapter (ACP) has the adapter as its +// parent, and an orphan's parent is the subreaper, which may be pid 1. A pid +// this harness did not record is never signaled. +func (w *fakeWorker) killConnector(ctx context.Context) error { + running, err := harnessConnector(w.dir) + if err != nil { + return err + } + pid := running.PID + // Only while it is still the process that wrote the file: a pid is not an + // identity. + if owns, err := driver.OwnsWorker(running); err != nil || !owns { + if err == nil { + err = errors.New("its start time no longer matches") + } + return fmt.Errorf("the connector's pid %d is no longer the connector: %w", pid, err) + } + if err := syscall.Kill(pid, syscall.SIGKILL); err != nil { + return fmt.Errorf("kill the connector (pid %d): %w", pid, err) + } + return waitFor(ctx, func() (bool, error) { return processGone(ctx, pid), nil }) +} + +// harnessConnector reads the identity the connector wrote when it started. +func harnessConnector(dir string) (driver.Process, error) { + data, err := os.ReadFile(filepath.Join(dir, connectorFile)) + if err != nil { + return driver.Process{}, err + } + var running struct { + PID int `json:"pid"` + StartedAt time.Time `json:"started_at"` + } + if err := json.Unmarshal(data, &running); err != nil { + return driver.Process{}, err + } + if running.PID <= 1 || running.StartedAt.IsZero() { + return driver.Process{}, fmt.Errorf("the connector recorded pid %d, which is nothing this harness may signal", running.PID) + } + // The group is only asked about when the process is gone; the kill is of + // the pid alone. + return driver.Process{PID: running.PID, PGID: running.PID, StartedAt: running.StartedAt}, nil +} + +func waitFor(ctx context.Context, cond func() (bool, error)) error { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + for { + ok, err := cond() + if err != nil { + return err + } + if ok { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(5 * time.Millisecond): + } + } +}