diff --git a/cmd/codeaf/app.go b/cmd/codeaf/app.go index ee268f0530..4df794b4ca 100644 --- a/cmd/codeaf/app.go +++ b/cmd/codeaf/app.go @@ -116,6 +116,11 @@ func runApp(args []string) error { wire := os.Stdout os.Stdout = os.Stderr defer func() { os.Stdout = wire }() + // THE APP MAY GO BEFORE THIS ENGINE DOES. A quit does not wait for the + // wrap-up below, and a crash cannot: either way the pipes to the app close + // while the engine still writes things down. A write to a closed stdout or + // stderr must fail, not end the process mid-wrap-up (appleave.go). + surviveBrokenPipes() // THE ONE ROAD OUT (internal/leave): a first SIGTERM, SIGINT or SIGHUP // starts the quit's careful wrap-up, and a second ends the process at once. @@ -844,8 +849,12 @@ func appMint(workspace string) (string, error) { } // appAgentID is a conversation's id from its transcript: the folder's name. +// A cell keeps its transcript one level down (/.cell/transcript.jsonl), so +// the folder is session.DirOf's, never the transcript's parent: that parent is +// ".cell" for every cell conversation, and one id for all of them would make +// each new conversation answer for the others. func appAgentID(transcript string) string { - return filepath.Base(filepath.Dir(filepath.Clean(transcript))) + return filepath.Base(session.DirOf(filepath.Clean(transcript))) } // locked runs fn holding b.mu. Every critical section in this file goes diff --git a/cmd/codeaf/appagentid_cells_test.go b/cmd/codeaf/appagentid_cells_test.go new file mode 100644 index 0000000000..56bcab5d11 --- /dev/null +++ b/cmd/codeaf/appagentid_cells_test.go @@ -0,0 +1,108 @@ +package main + +import ( + "path/filepath" + "strings" + "testing" + + "github.com/Agent-Field/codeaf/internal/appwire" + "github.com/Agent-Field/codeaf/internal/cell" + "github.com/Agent-Field/codeaf/internal/factory" + "github.com/Agent-Field/codeaf/internal/session" +) + +// A CONVERSATION'S ID IS ITS FOLDER'S NAME, IN EITHER LAYOUT. A cell keeps its +// transcript at /.cell/transcript.jsonl and a legacy folder at +// /transcript.jsonl; both answer . A cell's transcript parent (".cell") +// is never an id: one id for every cell conversation made each new one answer +// for all the others. +func TestAppAgentIDIsTheFolderInBothLayouts(t *testing.T) { + bucket := t.TempDir() + for _, tc := range []struct{ transcript, want string }{ + {filepath.Join(bucket, "5e55a0f1", cell.TranscriptPath), "5e55a0f1"}, + {filepath.Join(bucket, "6f66b1a2", "transcript.jsonl"), "6f66b1a2"}, + } { + if got := appAgentID(tc.transcript); got != tc.want { + t.Errorf("appAgentID(%s) = %q, want %q", tc.transcript, got, tc.want) + } + } +} + +// TWO NEW CONVERSATIONS ARE TWO IDS WHEN CELLS ARE ON. Each create answers its +// own folder's name, never ".cell", and each id finds its own conversation. +func TestAppCreateGivesEachCellConversationItsFolderName(t *testing.T) { + defer cell.UseAvailabilityInTests(true)() + useLocalHome(t) + rec := newAppRecorder() + backend := newFakeBackend(rec, func(ask appOpenAsk) (*appConn, error) { + return &appConn{Conv: &fakeConversation{}, Workspace: ask.Workspace, Transcript: ask.Transcript, Release: func() {}}, nil + }) + // The real mint: each create gets a folder of its own under this test's home. + backend.mint = appMint + backend.state = &appState{path: filepath.Join(t.TempDir(), appStateName)} + defer backend.Close() + ws := t.TempDir() + one, err := backend.CreateAgent(appwire.CreateAsk{Workspace: ws}) + if err != nil { + t.Fatal(err) + } + two, err := backend.CreateAgent(appwire.CreateAsk{Workspace: ws}) + if err != nil { + t.Fatal(err) + } + if one.ID == two.ID || one.ID == cell.StateDir || two.ID == cell.StateDir { + t.Fatalf("two creates answered %q and %q", one.ID, two.ID) + } + for _, id := range []string{one.ID, two.ID} { + a, err := backend.find(id) + if err != nil { + t.Fatalf("%s does not find its conversation: %v", id, err) + } + if filepath.Base(filepath.Dir(a.transcript)) != cell.StateDir || filepath.Base(session.DirOf(a.transcript)) != id { + t.Fatalf("%s is not its own cell folder: %s", id, a.transcript) + } + // Its media are its own folder's: listed under its id, kept in its folder, never in ".cell". + scope, roots, err := backend.mediaScope(id) + if err != nil { + t.Fatal(err) + } + folder := session.DirOf(a.transcript) + if scope != id { + t.Fatalf("%s's media are listed under %q", id, scope) + } + for _, root := range roots[1:] { + if !strings.HasPrefix(root, folder+string(filepath.Separator)) || strings.Contains(root, string(filepath.Separator)+cell.StateDir+string(filepath.Separator)+cell.StateDir) { + t.Fatalf("%s's media root %s is not under its folder %s", id, root, folder) + } + } + } +} + +// A FACTORY JOB'S CHATS NAME ITS TALK'S FOLDER WHEN THE TALK IS A CELL. +func TestAJobsChatsNameTheCellTalksFolder(t *testing.T) { + defer cell.UseAvailabilityInTests(true)() + s := newLocalSetup(t) + s.backend.open = func(ask appOpenAsk) (*appConn, error) { + return &appConn{Conv: &fakeConversation{}, Workspace: ask.Workspace, Transcript: ask.Transcript, Release: func() {}}, nil + } + st := s.start() + it, err := s.backend.local.host.floor.Create(factory.Item{Repo: "acme/api", Num: 8, Title: "Fix the ledger", Kind: factory.KindIssue, State: factory.StateNew}) + if err != nil { + t.Fatal(err) + } + var chats appwire.FactoryJobChats + s.must("factory.job_chats", map[string]any{"factory": st.Factory, "job": it.ID}, &chats) + kept, err := s.backend.local.host.floor.Get(it.ID) + if err != nil { + t.Fatal(err) + } + if filepath.Base(filepath.Dir(kept.Talk)) != cell.StateDir { + t.Fatalf("the job's talk is not a cell: %s", kept.Talk) + } + if want := filepath.Base(session.DirOf(kept.Talk)); chats.Agent != want || chats.Agent == cell.StateDir { + t.Fatalf("job chats answered agent %q, want the talk's folder %q", chats.Agent, want) + } + if !s.backend.isHeld(chats.Agent) { + t.Fatal("the job's agent is not held open under its folder's id") + } +} diff --git a/cmd/codeaf/appconvo.go b/cmd/codeaf/appconvo.go index a47d5aee8b..23428a0fcc 100644 --- a/cmd/codeaf/appconvo.go +++ b/cmd/codeaf/appconvo.go @@ -1916,7 +1916,7 @@ func appTrimmedStart(path string) int { func (b *appBackend) recordAttachedMedia(a *appAgent, attachments []appwire.Attachment) { var workspace string b.locked(func() { workspace = a.view.Workspace }) - id := filepath.Base(filepath.Dir(a.transcript)) + id := appAgentID(a.transcript) for _, attach := range attachments { if attach.Kind == appwire.AttachFile { if source := session.RecordAttachment(appMediaIndex(), id, appAttachFull(workspace, attach.Path)); source != "" { diff --git a/cmd/codeaf/appdeliverables.go b/cmd/codeaf/appdeliverables.go index bb4255f2b0..6216b176e9 100644 --- a/cmd/codeaf/appdeliverables.go +++ b/cmd/codeaf/appdeliverables.go @@ -8,7 +8,6 @@ import ( "errors" "io" "os" - "path/filepath" "strings" "time" "unicode/utf8" @@ -77,7 +76,7 @@ func (e *appDeliverEnv) agent(id string) *appDeliverAgent { if a, err := e.b.find(id); err == nil { var workspace string e.b.locked(func() { workspace = a.view.Workspace }) - place := session.Place{Dir: filepath.Dir(a.transcript), Owned: true} + place := session.Place{Dir: session.DirOf(a.transcript), Owned: true} out = &appDeliverAgent{ roots: []string{workspace, place.Work(), place.Artifacts()}, folders: []string{place.Work(), place.Artifacts()}, diff --git a/cmd/codeaf/applazy.go b/cmd/codeaf/applazy.go index b84010c18b..9dba733b2a 100644 --- a/cmd/codeaf/applazy.go +++ b/cmd/codeaf/applazy.go @@ -141,7 +141,7 @@ func (b *appBackend) lazyCandidates(run *appCrewRun, read appCrewRead, first boo if a.listening.IsZero() { a.listening = run.since } - candidates[id] = session.SessionRow{ID: id, Dir: filepath.Dir(a.transcript), Transcript: a.transcript} + candidates[id] = session.SessionRow{ID: id, Dir: session.DirOf(a.transcript), Transcript: a.transcript} } }) return candidates diff --git a/cmd/codeaf/appleave.go b/cmd/codeaf/appleave.go index 8e08a2052a..0a7b9954d1 100644 --- a/cmd/codeaf/appleave.go +++ b/cmd/codeaf/appleave.go @@ -28,6 +28,8 @@ import ( "context" "fmt" "os" + "os/signal" + "syscall" "time" ) @@ -96,3 +98,13 @@ func appLeaveNow() { fmt.Fprintf(os.Stderr, "codeaf app: still leaving after %s; ending now\n", appLeaveWait) os.Exit(1) } + +// surviveBrokenPipes makes a write to a pipe whose reader has gone fail with +// EPIPE instead of ending the process. Go ends a program that writes to a +// broken stdout or stderr unless it asked for SIGPIPE; the app's pipes are +// both, and the engine may outlive the app (a crash, or a quit's wrap-up). +// It asks with Notify, never Ignore: an ignored SIGPIPE is inherited across +// exec, so every shell, tool and server this engine starts would lose it +// (`producer | head` would never end). The channel is never read; Notify +// drops a signal its channel has no room for. +func surviveBrokenPipes() { signal.Notify(make(chan os.Signal, 1), syscall.SIGPIPE) } diff --git a/cmd/codeaf/appleave_pipes_test.go b/cmd/codeaf/appleave_pipes_test.go new file mode 100644 index 0000000000..65aebcb779 --- /dev/null +++ b/cmd/codeaf/appleave_pipes_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +// THE ENGINE OUTLIVES ITS APP'S PIPES, AND ITS CHILDREN KEEP SIGPIPE. When +// the app has closed both pipes the engine's writes fail and it goes on to the +// end of its wrap-up; a shell it starts still ends a `producer | head` pipe. +func TestAnEngineWritesOnAfterItsAppsPipesClose(t *testing.T) { + if dir := os.Getenv("CODEAF_TEST_PIPES_CHILD"); dir != "" { + surviveBrokenPipes() + // The parent closes its ends, then says so: every write after this is + // to a broken pipe, never into a buffer someone still reads. + deadline := time.Now().Add(10 * time.Second) + for _, err := os.Stat(filepath.Join(dir, "closed")); err != nil; _, err = os.Stat(filepath.Join(dir, "closed")) { + if time.Now().After(deadline) { + os.Exit(3) + } + time.Sleep(10 * time.Millisecond) + } + var outErr, errErr error + for outErr == nil || errErr == nil { + if time.Now().After(deadline) { + os.Exit(4) + } + if outErr == nil { + _, outErr = os.Stdout.WriteString("an event the app will never read\n") + } + if errErr == nil { + _, errErr = os.Stderr.WriteString("a log line no one collects\n") + } + } + if !errors.Is(outErr, syscall.EPIPE) || !errors.Is(errErr, syscall.EPIPE) { + os.Exit(5) + } + // A shell this engine starts gets SIGPIPE back: its producer ends at once. + status, err := exec.Command("bash", "-c", "yes | head -1 >/dev/null; echo ${PIPESTATUS[0]}").Output() + if err != nil { + os.Exit(6) + } + _ = os.WriteFile(filepath.Join(dir, "child-sigpipe"), status, 0o600) + _ = os.WriteFile(filepath.Join(dir, "wrapped-up"), []byte("ok"), 0o600) + os.Exit(0) + } + if _, err := exec.LookPath("bash"); err != nil { + t.Skip("no bash to start a child shell with") + } + dir := t.TempDir() + cmd := exec.Command(os.Args[0], "-test.run=^TestAnEngineWritesOnAfterItsAppsPipesClose$") + cmd.Env = append(os.Environ(), "CODEAF_TEST_PIPES_CHILD="+dir) + out, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + errs, err := cmd.StderrPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + _ = out.Close() + _ = errs.Close() + if err := os.WriteFile(filepath.Join(dir, "closed"), nil, 0o600); err != nil { + t.Fatal(err) + } + if err := cmd.Wait(); err != nil { + t.Fatalf("the engine ended while writing to its gone app: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "wrapped-up")); err != nil { + t.Fatalf("the wrap-up did not finish: %v", err) + } + status, _ := os.ReadFile(filepath.Join(dir, "child-sigpipe")) + if got := strings.TrimSpace(string(status)); got != "141" { + t.Fatalf("a shell the engine starts ended its producer with %q, want 141 (SIGPIPE)", got) + } +} diff --git a/cmd/codeaf/appmedia.go b/cmd/codeaf/appmedia.go index e5e5e0d48b..7b2f8b7db8 100644 --- a/cmd/codeaf/appmedia.go +++ b/cmd/codeaf/appmedia.go @@ -31,7 +31,8 @@ func (b *appBackend) mediaScope(agent string) (sessionID string, roots []string, } var workspace string b.locked(func() { workspace = a.view.Workspace }) - dir := filepath.Dir(a.transcript) + // The session folder, in either layout: a cell keeps its transcript in .cell (appAgentID). + dir := session.DirOf(a.transcript) place := session.Place{Dir: dir, Owned: true} roots = []string{workspace, place.Work(), place.Artifacts()} return filepath.Base(dir), roots, nil diff --git a/cmd/codeaf/appmedia_attach.go b/cmd/codeaf/appmedia_attach.go index 4a654458c2..805dd653ed 100644 --- a/cmd/codeaf/appmedia_attach.go +++ b/cmd/codeaf/appmedia_attach.go @@ -106,7 +106,7 @@ func (b *appBackend) attachReference(agent, source string, span appSpan, snapsho return appwire.Attachment{}, appMediaError(err) } // Owned session storage, never the original's directory or the workspace. - destination := (session.Place{Dir: filepath.Dir(a.transcript), Owned: true}).Artifacts() + destination := (session.Place{Dir: session.DirOf(a.transcript), Owned: true}).Artifacts() if err := os.MkdirAll(destination, 0o700); err != nil { return appwire.Attachment{}, err } diff --git a/docs/changes/unreleased/1973-app-cell-ids-and-broken-pipes.md b/docs/changes/unreleased/1973-app-cell-ids-and-broken-pipes.md new file mode 100644 index 0000000000..a6c3088224 --- /dev/null +++ b/docs/changes/unreleased/1973-app-cell-ids-and-broken-pipes.md @@ -0,0 +1,16 @@ +--- +kind: fixed +title: A conversation's id is its folder's name, and an app engine survives its app's pipes closing +pr: 1973 +surface: [engine] +invalidates: + - "Was true: the app protocol named every cell conversation \".cell\", so a factory job's chats were not found, a newly created agent answered for every job's chat, and media of one cell conversation were listed under all of them. Is true: a conversation's id is its session folder's name in both layouts, and its media are its own." + - "Was true: an app engine ended with SIGPIPE if it wrote to stdout or stderr after its app had closed the pipes, cutting its leaving wrap-up short. Is true: those writes fail and the wrap-up finishes, while the shells and tools it starts keep SIGPIPE as usual." +--- + +Since cell conversations keep their transcript in `/.cell/`, the app +protocol took `.cell` for every id. Ids, media listings, attachment staging, +deliverables and lazy wake-ups now use the session folder. `codeaf app` asks +for SIGPIPE with `signal.Notify` rather than ignoring it, so a write to a +closed app pipe fails instead of ending the engine, and nothing it starts +inherits an ignored SIGPIPE.