Join a headless run's late accounting before it returns, and isolate the shared team fixtures

cmd/codeaf/chat.go

@@ -955,6 +955,7 @@ func buildBrain(w *chatWindow, session string, opts brainOptions) (*chatBrain, e
 		// Every call this leaf makes, banked under its node as the provider
 		// answers. See leafBanker.
 		banker := newLeafBanker(graph, node.ID)
+		banker.receipts = &brain.background
 		var spent exec.Usage
 		// The same spend with its per-turn shape kept, accumulated across every
 		// attempt this node makes. See leafShape.
@@ -2673,10 +2674,14 @@ func atLeastZero(count int) int {
 // and a leaf whose calls were banked here is a leaf whose total is already on
 // disk (see resident.ExecResult.SpendBanked).
 type leafBanker struct {
-	graph  *store.Store
-	nodeID string
-	mutex  sync.Mutex
-	rows   int
+	usageLedger string
+	// Cancelled arms can still owe receipts after the worker lands. Join them
+	// within the brain's existing background stop bound before closing its store.
+	receipts *sync.WaitGroup
+	graph    *store.Store
+	nodeID   string
+	mutex    sync.Mutex
+	rows     int
 	// total is what this banker has actually written, so the landing can journal
 	// the REMAINDER rather than the whole — see leafSpend.
 	total exec.Usage
@@ -2687,7 +2692,7 @@ type leafBanker struct {
 }
 
 func newLeafBanker(graph *store.Store, nodeID string) *leafBanker {
-	return &leafBanker{graph: graph, nodeID: nodeID}
+	return &leafBanker{graph: graph, nodeID: nodeID, usageLedger: session.UsageLedgerPath()}
 }
 
 // bank writes one call's row. It is called from the provider's decode, which is
@@ -2729,7 +2734,7 @@ func (b *leafBanker) reconciled(receipt provider.Reconciled) {
 		return
 	}
 	if !receipt.Found {
-		session.RecordUnbilledCall(session.UsageLedgerPath(), session.UsageLine{Task: b.nodeID, Model: receipt.Model})
+		session.RecordUnbilledCall(b.usageLedger, session.UsageLine{Task: b.nodeID, Model: receipt.Model})
 		return
 	}
 	b.bank(receipt.Billed)
@@ -2756,6 +2761,12 @@ func armBilling(ctx context.Context, banker *leafBanker) context.Context {
 		return ctx
 	}
 	ctx = provider.WithCallNode(ctx, banker.nodeID)
+	if banker.receipts != nil {
+		ctx = provider.WithReceiptSettlement(ctx, func() func() {
+			banker.receipts.Add(1)
+			return banker.receipts.Done
+		})
+	}
 	ctx = provider.WithBilling(ctx, banker.bank)
 	return provider.WithReconcile(ctx, banker.reconciled)
 }

cmd/codeaf/do.go

@@ -572,6 +572,8 @@ func applyContextLaw(fillPercent, completionReserve int) error {
 // to main, which says it on stderr — an ordinary run has no object to put it in
 // and never wanted one.
 func doErrand(request doRequest) error {
+	// Producers stop before the process-owned usage writers are joined.
+	defer session.CloseUsage()
 	started := time.Now()
 	// The two seats, resolved before anything is opened or built, so the run
 	// says which models it is about to use and on whose authority — and says it

cmd/codeaf/keptnothing_test.go

@@ -12,6 +12,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/Agent-Field/codeaf/internal/session"
 )
 
 // TestARefusedRunThatWroteNothingNamesTheDirectory proves C1: a refused run
@@ -341,3 +343,37 @@ func TestAPriceRefusalDoesNotClaimTheRunEditedAWorkspace(t *testing.T) {
 		t.Fatalf("a price refusal claims it worked in the directory: %q", got)
 	}
 }
+
+// A cancelled provider arm can owe an unbilled marker after the worker lands.
+// The command owns both that producer and the writer it starts.
+func TestDoReturnsAfterCancelledAccountingAndItsWriterStop(t *testing.T) {
+	session.CloseUsage()
+	t.Setenv("CODEAF_HOME", t.TempDir())
+	script := newScriptedBrain(t)
+	script.stall = true
+	defer script.close()
+	t.Cleanup(session.CloseUsage)
+	var stdout, stderr strings.Builder
+	err := doErrand(doRequest{task: "write the release note and include migration steps", workspace: t.TempDir(),
+		timeout: 2 * time.Second, stdout: &stdout, stderr: &stderr, newClient: script.client})
+	var status exitStatus
+	if !asExitStatus(err, &status) || status != exitLimit {
+		t.Fatalf("timeout exit = %v; stderr %s", err, stderr.String())
+	}
+	rows, err := session.ReadUsage(session.UsageLedgerPath(), time.Time{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, row := range rows {
+		if row.Unbilled && row.Task != "" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("command returned before cancelled-call accounting: %+v", rows)
+	}
+	if n := liveUsageWriters(); n != 0 {
+		t.Fatalf("command returned with %d usage writers still running", n)
+	}
+}

docs/changes/unreleased/2604-post-wave-test-isolation.md

@@ -0,0 +1,19 @@
+---
+kind: fixed
+title: Join cancelled headless accounting before returning and isolate shared team fixtures
+pr: 2604
+surface: [chat, engine, build, docs]
+invalidates:
+  - "A local headless task could return before cancelled-call accounting created its usage writer; it now joins that producer before closing usage."
+---
+
+The node engine joins admitted leaf receipt settlement within its existing
+background stop wait. Every `codeaf do` invocation closes the process usage writers
+after its workers stop, under the existing usage close bound. Each leaf's late
+accounting retains the ledger chosen when the leaf was constructed.
+Cancelled requests still keep their unbilled markers, and journal sync and retry
+retain their existing durability and shutdown limits.
+
+Shared team fixtures create a paired identity in a test-owned home instead of
+inheriting a solo identity from an earlier custody save. Explicit solo delivery
+and device cursor assertions retain their original postures.

internal/manual/chat/adaptive-runs.md

@@ -983,6 +983,19 @@ no turns at all is not offered again either, because there is nothing for a next
 carry on from, and it says that instead: `…with none of its work recorded, so there was
 nothing for another attempt to carry on from`.
 
+## What do and exec wait for at shutdown — usage writes after a timeout
+
+`codeaf do` closes the process's usage writers after its workers stop, including
+when the run times out. With `CODEAF_TASK_BELT=node`, cancelled workers' admitted
+receipt accounting joins inside the existing five-second background stop wait.
+Usage shutdown has a separate two-second bound. With a responsive provider and
+disk, the command returns after those accounting writes finish. A stalled
+provider or disk can exhaust a bound; complete rows already in the pending usage
+journal remain recoverable on restart.
+
+`codeaf exec` also closes its usage writer before returning, under the same
+two-second usage shutdown bound.
+
 ## How long a headless run gets — the `-timeout` wall, how long does codeaf do wait, can I write 5m or 2h
 
 `codeaf do` runs under a hard wall, and `-timeout` is where you set it. **Left alone it is

internal/session/team_locality_test.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/Agent-Field/codeaf/internal/config"
 	"github.com/Agent-Field/codeaf/internal/identity"
 	"github.com/Agent-Field/codeaf/internal/teams"
 )
@@ -20,7 +21,11 @@ func TestLocalTeamDeliverySurvivesMissingFilesAndLookupFailure(t *testing.T) {
 		t.Fatal(err)
 	}
 	f := newTeamFixture(t, true)
-	enableSessionTeamSharing(t, f.profile)
+	// This fixture deliberately remains solo with sharing requested.
+	row, _ := config.NewSettings(config.SettingsOptions{ProfileDir: f.profile}).Row(config.KeyTeamsShare)
+	if err := row.Apply("on"); err != nil {
+		t.Fatal(err)
+	}
 	a := teamAgent(t, f, f.manager, nil, manualRelayTeamClock)
 	a.config.TeamComputers = func(context.Context, []teams.Member) (map[string]string, error) { return nil, errors.New("offline") }
 	if err := teams.Update(f.profile, func(file *teams.File) error {

internal/session/team_relay_test.go

@@ -88,17 +88,13 @@ func TestTrafficMirrorNeverDeliversStartsQuestionsWakesOrStops(t *testing.T) {
 }
 
 func TestForeignDeliveryCursorStartsAtThisComputersOpeningAndDeliversOnce(t *testing.T) {
-	root := t.TempDir()
-	t.Setenv("CODEAF_HOME", root)
-	if _, err := identity.Ensure(root); err != nil {
-		t.Fatal(err)
-	}
+	f := newWakingTeamFixture(t)
+	enableSessionTeamSharing(t, f.profile)
+	root := home.Dir()
 	dev, err := identity.Device(root)
 	if err != nil {
 		t.Fatal(err)
 	}
-	f := newWakingTeamFixture(t)
-	enableSessionTeamSharing(t, f.profile)
 	// A line from long before the chat arrived is history; the directive sent
 	// a second before the engine opened is the one a rouse opened it for, which
 	// a conversation meeting a team for the first time is always handed.
@@ -207,12 +203,12 @@ func TestRemoteTeamSendsStartsPostsAndRousesRefuseBeforeTraffic(t *testing.T) {
 
 func enableSessionTeamSharing(t *testing.T, profile string) {
 	t.Helper()
-	if _, err := identity.Load(home.Dir()); err != nil {
-		root := t.TempDir()
-		t.Setenv("CODEAF_HOME", root)
-		if _, err := identity.Ensure(root); err != nil {
-			t.Fatal(err)
-		}
+	// Custody saves in other fixtures can leave the process home solo.
+	// Sharing needs its own paired home, restored after this test's agents close.
+	root := t.TempDir()
+	t.Setenv("CODEAF_HOME", root)
+	if _, err := identity.Ensure(root); err != nil {
+		t.Fatal(err)
 	}
 	row, _ := config.NewSettings(config.SettingsOptions{ProfileDir: profile}).Row(config.KeyTeamsShare)
 	if err := row.Apply("on"); err != nil {