tui3, chat, pool: task-page keys, notes, tab mark and unowned warms

cmd/codeaf/chatv3.go

@@ -890,7 +890,14 @@ func openV3Launch(proc *v3Process, opts v3Options) (*v3Launch, error) {
 	// has a good answer without it, and an unknown window leaves the session on
 	// its conservative default.
 	models := proc.Models
-	activeModels, activeModel, activeListsModels := v3CatalogForModel(context.Background(), settings, chosen, models)
+	// A DIRECT SERVICE'S OWN CATALOG IS A WARM OF ITS OWN, and it is owned the
+	// way the process's is (#1274): it runs under the process lifetime and
+	// closeAll cancels and joins it, so no cache write can land after the
+	// process has closed or in a later home.
+	activeModels, activeModel, activeListsModels := v3CatalogForModel(proc.lifetime(), settings, chosen, models)
+	if activeModels != models {
+		proc.ownCatalog(activeModels)
+	}
 	harnesses := proc.Harnesses
 
 	// The typed programs this conversation can reach, and the two stores they are

cmd/codeaf/chatv3_beside.go

@@ -144,6 +144,31 @@ type engineFleet struct {
 
 	mu    sync.Mutex
 	conns []*engineConn
+	// closers are what this window opened beside its connections and must join
+	// when it lets them go — the model catalog [hostOptions] warms, whose warm
+	// writes a cache when it lands (#1274). [engineFleet.closeAll] runs them.
+	closers []func()
+}
+
+// own hands one close to [engineFleet.closeAll]. A nil fleet or a nil close is
+// nothing to own.
+func (f *engineFleet) own(shut func()) {
+	if f == nil || shut == nil {
+		return
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.closers = append(f.closers, shut)
+}
+
+// takeClosers hands over everything [engineFleet.own] was given and empties the
+// list, so the joins run after the lock is let go.
+func (f *engineFleet) takeClosers() []func() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	closers := f.closers
+	f.closers = nil
+	return closers
 }
 
 // machineReadings is the handful of per-window facts a beside conversation needs
@@ -303,6 +328,11 @@ func (f *engineFleet) closeAll() {
 		}
 	}
 	_ = f.boot.close()
+	// AND WHAT THE WINDOW OPENED BESIDE THEM, joined last: nothing it warmed may
+	// write after the door has let go.
+	for _, shut := range f.takeClosers() {
+		shut()
+	}
 }
 
 // takeAll hands over every connection this window opened and empties the list,

cmd/codeaf/chatv3_guard_class_test.go

@@ -15,9 +15,9 @@ import (
 // v3 process or launch. A long-lived writer must identify both its stop and the
 // closeAll join. chatv3/models and engine/models name the process waiter, joined
 // since #1179, and the internal/catalog warm each one starts, which closeAll now
-// cancels and joins through Models.Close.
-// The remaining known-open row (chatv3/sweep-home) names the later cell that owns it;
-// deleting a row is part of landing that cell.
+// cancels and joins through Models.Close. chatv3/sweep-home is joined too, by
+// stopPlaceSweep since #1276. No row is known-open now; a new one that is must
+// name the later cell that owns it, and deleting it is part of landing that cell.
 func TestV3ProcessGuardGoClass(t *testing.T) {
 	type class struct {
 		kind, stop, join, owner string
@@ -28,7 +28,7 @@ func TestV3ProcessGuardGoClass(t *testing.T) {
 		"pool/judge-sweep":        {kind: "joined writer", stop: "pool errand context", join: "v3Process.closeAll calls stopPoolErrands"},
 		"chatv3/models":           {kind: "joined writer", stop: "warmModels waiter on the pool errand context, and the catalog's own warm context", join: "v3Process.closeAll calls stopPoolErrands for the waiter (since #1179) and Models.Close, which cancels and joins the internal/catalog warm"},
 		"engine/models":           {kind: "joined writer", stop: "warmModels waiter on the pool errand context, and the catalog's own warm context", join: "v3Process.closeAll calls stopPoolErrands for the waiter (since #1179) and Models.Close, which cancels and joins the internal/catalog warm"},
-		"chatv3/sweep-home":       {kind: "KNOWN-OPEN writer", owner: "later home-sweep ownership cell (class target three)"},
+		"chatv3/sweep-home":       {kind: "joined writer", stop: "v3Process.sweepCancel cancels the walk's context, checked between entries and before every destructive operation", join: "v3Process.closeAll calls stopPlaceSweep, which waits on sweepDone (since #1276)"},
 		"chatv3/background":       {kind: "one-shot", stop: "repairBackgroundChecks returns after one bounded Drift/Install pass"},
 		"chatv3/once-questions":   {kind: "one-shot", stop: "agent.Close closes the WatchQuestions channel consumed by the range"},
 		"chatv3/close-agent":      {kind: "joined one-shot", stop: "Agent.Close is bounded", join: "v3Process.closeAll waits on waiting"},

cmd/codeaf/chatv3_host.go

@@ -616,6 +616,10 @@ func hostOptions(fleet *engineFleet, welcome remote.Welcome, pick bool) (tui3.Op
 	}
 	discovery := catalog.Options{BaseURL: settings.BaseURL, APIKey: settings.APIKey, Dir: profileDir}
 	models := catalog.LoadLazy(context.Background(), discovery)
+	// THE WARM IS THE WINDOW'S, and the window's fleet joins it when the door
+	// lets go (#1274): its fetch writes a cache when it lands, and one nobody
+	// joined could write after this window had closed.
+	fleet.own(models.Close)
 	// A tier row that says auto is answered from this catalog (config.AutoModels):
 	// the same non-blocking read, never a fetch, and set once at start-up.
 	config.AutoModels = models.ModelsNow

cmd/codeaf/chatv3_process.go

@@ -113,7 +113,55 @@ type v3Process struct {
 	standingDone    chan struct{}
 	sweepCancel     context.CancelFunc
 	sweepDone       chan struct{}
-	closed          bool
+	// catalogs are the lazy catalogs this process opened beside Models — a
+	// direct service's own listing, asked for when a conversation is opened on
+	// one of its models ([v3Process.ownCatalog]). closeAll cancels and joins each.
+	catalogs []*catalog.Catalog
+	closed   bool
+}
+
+// lifetime is the context background work owned by this process runs under,
+// and it ends when closeAll begins. A process built without one (a test's bare
+// literal) hands out the plain background, which is what it had before.
+func (p *v3Process) lifetime() context.Context {
+	if p == nil || p.processCtx == nil {
+		return context.Background()
+	}
+	return p.processCtx
+}
+
+// ownCatalog hands a lazy catalog this process opened to closeAll, which
+// cancels and joins its warm the way it does Models' (#1274). A catalog handed
+// over after the close has begun is closed at once, so none is left unowned.
+func (p *v3Process) ownCatalog(models *catalog.Catalog) {
+	if p == nil || models == nil {
+		return
+	}
+	if !p.keepCatalog(models) {
+		models.Close()
+	}
+}
+
+// keepCatalog files one catalog for closeAll, and answers false when the close
+// has already begun and nothing will come back for it.
+func (p *v3Process) keepCatalog(models *catalog.Catalog) bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed {
+		return false
+	}
+	p.catalogs = append(p.catalogs, models)
+	return true
+}
+
+// takeCatalogs hands over every catalog [v3Process.ownCatalog] was given and
+// empties the list, so the joins run after the lock is let go.
+func (p *v3Process) takeCatalogs() []*catalog.Catalog {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	owned := p.catalogs
+	p.catalogs = nil
+	return owned
 }
 
 // openV3Process builds the once-only half of a v3 launch.
@@ -429,6 +477,11 @@ func (p *v3Process) closeAll() {
 	if p.Models != nil {
 		p.Models.Close()
 	}
+	// AND EVERY OTHER CATALOG THIS PROCESS OPENED, for the same promise: a direct
+	// service's own listing warms under the same lifetime and is joined here.
+	for _, models := range p.takeCatalogs() {
+		models.Close()
+	}
 
 	// Cancellation is checked between entries and before destructive operations.
 	// Joining therefore waits only for the current bounded filesystem operation,

cmd/codeaf/chatv3_standing.go

@@ -121,17 +121,22 @@ func standingWatch(store *standing.Store) standing.Watch {
 // where a machine that has none says so on that item's own row — `could not
 // check: no API key: this session has not been given one yet` — and the rest of
 // the walk goes on.
-func v3StandingTicker(store *standing.Store) (*standing.Ticker, error) {
+//
+// IT ANSWERS A RELEASE BESIDE THE PASS, which the caller runs when the pass is
+// over: the posture's model catalog warms in the background and writes a cache
+// when it lands, and the release cancels and joins it (#1274). It is never nil
+// when the error is.
+func v3StandingTicker(store *standing.Store) (*standing.Ticker, func(), error) {
 	if store == nil {
-		return nil, fmt.Errorf("standing: no store")
+		return nil, nil, fmt.Errorf("standing: no store")
 	}
 	settings, err := config.LoadKeyless()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	posture, err := v3StandingPosture(settings)
+	posture, models, err := v3StandingPosture(settings)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	idle := session.StandingIdle()
 	return &standing.Ticker{
@@ -149,7 +154,7 @@ func v3StandingTicker(store *standing.Store) (*standing.Ticker, error) {
 		// seam nil and the pass absent.
 		Tidy:         session.NewMemoryTidy(posture, v3MemoryPath(settings.ProfileDir), store.Root(), idle),
 		DailyRailUSD: v3StandingDailyRail(settings.ProfileDir),
-	}, nil
+	}, models.Close, nil
 }
 
 // v3MemoryPath is the brain's file when the memory row is on, and the empty
@@ -172,7 +177,10 @@ func v3MemoryPath(profileDir string) string {
 //
 // The workspace here is only where the rows are read from. Every firing runs in
 // its own item's workspace, which the runner sets before it opens anything.
-func v3StandingPosture(settings config.Config) (session.Config, error) {
+//
+// It answers the lazy catalog it opened as well, which the caller closes when
+// the pass is over; closing it does not take the rows it has already read.
+func v3StandingPosture(settings config.Config) (session.Config, *catalog.Catalog, error) {
 	root, err := os.UserHomeDir()
 	if err != nil || root == "" {
 		root = os.TempDir()
@@ -206,13 +214,18 @@ func v3StandingPosture(settings config.Config) (session.Config, error) {
 	// have.
 	cfg, err = applyV3Governance(cfg, settings.ProfileDir, false, false)
 	if err != nil {
-		return session.Config{}, err
+		return session.Config{}, nil, err
 	}
 	// The media pair, resolved the way a conversation resolves it (chatv3.go):
 	// a firing briefed to draw a diagram needs the hand that draws it, and the
 	// resolver is what says which model does. The catalog is LAZY and is never
 	// waited for — a pass whose catalog has not resolved simply has no media
 	// verbs on its belt, which is the same absence a cold conversation has.
+	//
+	// THE CATALOG IS THE PASS'S TO CLOSE (#1274). Its warm writes a cache when it
+	// lands, and a pass is rebuilt every five minutes; a warm nobody joined could
+	// write after the window had closed. So it is handed back to the one caller,
+	// [v3StandingTicker], whose release closes it when the pass is over.
 	models := catalog.LoadLazy(context.Background(), catalog.Options{
 		BaseURL: settings.BaseURL, APIKey: settings.APIKey, Dir: settings.ProfileDir,
 	})
@@ -223,7 +236,7 @@ func v3StandingPosture(settings config.Config) (session.Config, error) {
 	cfg.NearestModels = v3NearestModels(models)
 	// AskConsent stays false and Standing stays nil: nobody is watching a
 	// firing, and nothing that fires may arm anything else.
-	return cfg, nil
+	return cfg, models, nil
 }
 
 // startStandingTicks runs a pass every [standing.Interval] for as long as this
@@ -327,11 +340,12 @@ func runStandingTick(ctx context.Context, store *standing.Store) {
 	// whole contribution to the ambient side, and a goroutine that unwound out
 	// of it would leave a window that looks like it is keeping watch and is not.
 	defer guard.Recover("standing tick")
-	pass, err := v3StandingTicker(store)
+	pass, release, err := v3StandingTicker(store)
 	if err != nil {
 		noteStanding("could not start a pass: " + err.Error())
 		return
 	}
+	defer release()
 	// The pass's ceiling is a CHILD of the caller's ctx, so closing the ticker
 	// (which cancels that ctx) ends an in-flight pass at once, and the 120s
 	// TickWindow stays the pass's own upper bound when nobody is quitting.

cmd/codeaf/poolrecord.go

@@ -582,6 +582,19 @@ func sweepPending(settings config.Config, profileDir, poolDir string, models fun
 func sweepPendingContext(ctx context.Context, settings config.Config, profileDir, poolDir string, models func() []catalog.Model, ask func(model string) judge.Ask, now func() time.Time, deadline time.Time) (judged, left int) {
 	claim := pendingPath(poolDir) + ".sweeping"
 	judged, left = sweepClaimContext(ctx, settings, profileDir, poolDir, claim, models, ask, now, deadline)
+	// A CLAIM THAT IS STILL THERE WAS NOT FINISHED, and the fresh file must not be
+	// renamed over it. The sweep removes a claim only when it reached every row;
+	// one a cancel or the deadline cut short keeps its unjudged rows for the next
+	// start, and renaming the pending file onto the same name replaced those rows
+	// with the new ones and lost them for good (#1267). The pending file waits
+	// where it is, and the next start takes the leftover first, as this one did.
+	// Its rows are waiting too, so they are counted among the ones left.
+	if _, err := os.Lstat(claim); err == nil {
+		if data, err := os.ReadFile(pendingPath(poolDir)); err == nil {
+			left += countUnjudged(poolDir, strings.Split(string(data), "\n"))
+		}
+		return judged, left
+	}
 	if err := os.Rename(pendingPath(poolDir), claim); err != nil {
 		return judged, left
 	}

cmd/codeaf/poolsweep_test.go

@@ -7,7 +7,10 @@ package main
 // pending file.
 
 import (
+	"context"
+	"encoding/json"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -125,3 +128,67 @@ func TestPoolJudgeSweepToleratesATornPendingLine(t *testing.T) {
 		t.Fatal("the torn row was judged; it should have been skipped")
 	}
 }
+
+// pendingIDs reads the landing ids a pending file or a claim holds, the way the
+// sweep reads them: one row a line, a torn line counted as nothing.
+func pendingIDs(t *testing.T, path string) map[uint64]bool {
+	t.Helper()
+	ids := map[uint64]bool{}
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return ids
+	}
+	for _, line := range strings.Split(string(data), "\n") {
+		var row pendingLanding
+		if json.Unmarshal([]byte(strings.TrimSpace(line)), &row) == nil {
+			ids[row.Landing.ID] = true
+		}
+	}
+	return ids
+}
+
+// TestPoolJudgeSweepCutShortKeepsALeftoverClaimsRows: a sweep that stops
+// before it has reached every row of a leftover claim leaves that claim for the
+// next start. The fresh pending file must not then be renamed over it, which
+// replaced the leftover's unjudged rows with the new ones and lost them for good.
+func TestPoolJudgeSweepCutShortKeepsALeftoverClaimsRows(t *testing.T) {
+	t.Setenv("CODEAF_HOME", t.TempDir())
+	t.Setenv("CODEAF_MODEL_POOL", "on")
+	t.Setenv("CODEAF_MODEL_POOL_SUBMIT_URL", "http://127.0.0.1:1/submit")
+	restoreOwnCells(t)
+
+	profileDir := t.TempDir()
+	poolDir := config.ProfilePath(profileDir, "pool")
+	settings := config.Config{APIKey: "k"}
+
+	// A LEFTOVER CLAIM, from a sweep a process death cut short.
+	leftover := poolTestLanding()
+	leftover.ID = 101
+	if err := writePendingLanding(profileDir, "do", leftover); err != nil {
+		t.Fatalf("write leftover row: %v", err)
+	}
+	claim := pendingPath(poolDir) + ".sweeping"
+	if err := os.Rename(pendingPath(poolDir), claim); err != nil {
+		t.Fatalf("leave the claim behind: %v", err)
+	}
+	// AND A ROW A DOOR WROTE SINCE.
+	fresh := poolTestLanding()
+	fresh.ID = 202
+	if err := writePendingLanding(profileDir, "do", fresh); err != nil {
+		t.Fatalf("write fresh row: %v", err)
+	}
+
+	// THE CLOSE HAS ALREADY CANCELLED THE SWEEP.
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var asked []string
+	sweepPendingContext(ctx, settings, profileDir, poolDir, poolTestCatalog, poolTestAsk(settings, &asked), time.Now, time.Now().Add(time.Minute))
+
+	inClaim, inPending := pendingIDs(t, claim), pendingIDs(t, pendingPath(poolDir))
+	if !inClaim[101] {
+		t.Fatalf("the leftover claim lost its unjudged row 101: the claim holds %v, pending holds %v", inClaim, inPending)
+	}
+	if !inClaim[202] && !inPending[202] {
+		t.Fatalf("the fresh row 202 is in neither file: the claim holds %v, pending holds %v", inClaim, inPending)
+	}
+}

cmd/codeaf/tick.go

@@ -32,10 +32,13 @@ func runTick(args []string) error {
 	if err != nil {
 		return err
 	}
-	ticker, err := v3StandingTicker(store)
+	ticker, release, err := v3StandingTicker(store)
 	if err != nil {
 		return err
 	}
+	// The pass's model catalog is joined before this command returns, so its
+	// warm cannot write a cache after the process has said it is done.
+	defer release()
 	ctx, cancel := context.WithTimeout(context.Background(), standing.TickWindow)
 	defer cancel()
 	if _, err := ticker.Tick(ctx); err != nil {

docs/changes/unreleased/1395-task-page-keys-and-tab-mark.md

@@ -0,0 +1,26 @@
+---
+kind: fixed
+title: the task page's keys, its note and its tab mark say one thing, and three catalog warms are owned
+pr: 1395
+surface: [chat, engine]
+invalidates:
+  - "The tab in front answered 'waiting on a person' from the surface alone, so a landed your call, the model's own blocking question or a sub-harness's question wore a ? on the tab beside it and lost it when the conversation came forward. Both sides now read the engine's NeedsPerson; the front asks it once per message on the loop, never per frame."
+  - "x and p acted on an ended task's page and row although the key line hides both: x raised Stop this task? over a finished run, and on a finished part drew the store's raw 'is already terminal'. On an ended page they are letters in the note, and on an ended row they are letters."
+  - "Keys typed while a run task's page was opening were replayed through the page's whole keyboard, so a note beginning with x cancelled a running part and a typed enter sent it. They now go into the note box only, and an enter typed in that gap is dropped."
+  - "A note's enter pressed twice sent it twice, and its reply replaced whichever page was open with the page it was sent from. The send is one send while in flight, and the reply lands only on its own page."
+  - "A restart judge sweep that stopped early on a leftover pending.jsonl.sweeping claim renamed the fresh pending file over it and lost the leftover's unjudged rows. The fresh file now waits for the next start."
+  - "The direct-service catalog a conversation opens, the catalog each standing pass opens and the host door's catalog warmed under a context nothing cancelled and were never joined. The process, the pass and the window's fleet now close each."
+  - "The manual said a run task's page names its folder once in the head. It does not, and has not since #1257: the folder is the run's own working copy, and step commands are drawn without the change into it."
+---
+The front tab caches the engine's answer in app.frontWaits after every message,
+beside the surface's own cards, and frontSignal reads the cache; the strip still
+asks no lock per frame. The keeper's watcher already asked the same predicate
+after every event of a held conversation.
+
+ctrl+o on a run task's page counts the brief's rows at the width the page draws
+it rather than the conversation's body width. The work tab draws a note's author
+the page's way, `you` or nothing. The first-run model chooser filters through
+internal/fuzzy, so `ds v4` finds what /model finds. The unread config keys notice
+is not a tip and shows with hints off; commands.md has its section, and
+screen.md says which fenced languages are coloured and that the rest are plain.
+The goroutine inventory's chatv3/sweep-home row is joined, as #1276 made it.