app: a conversation's id is its folder's name; finish the wrap-up after the app's pipes close

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 (<id>/.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

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 <id>/.cell/transcript.jsonl and a legacy folder at
+// <id>/transcript.jsonl; both answer <id>. 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")
+	}
+}

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 != "" {

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()},

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

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) }

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)
+	}
+}

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

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
 	}

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 `<id>/.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.