diff --git a/cmd/codeaf/chat.go b/cmd/codeaf/chat.go index f68ec2024e..4ebe20a906 100644 --- a/cmd/codeaf/chat.go +++ b/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) } diff --git a/cmd/codeaf/do.go b/cmd/codeaf/do.go index 88f9d5208b..a6421b07a8 100644 --- a/cmd/codeaf/do.go +++ b/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 diff --git a/cmd/codeaf/keptnothing_test.go b/cmd/codeaf/keptnothing_test.go index 1fe948d785..0257008c49 100644 --- a/cmd/codeaf/keptnothing_test.go +++ b/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) + } +} diff --git a/docs/changes/unreleased/2604-post-wave-test-isolation.md b/docs/changes/unreleased/2604-post-wave-test-isolation.md new file mode 100644 index 0000000000..7f22a6dfe7 --- /dev/null +++ b/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. diff --git a/internal/manual/chat/adaptive-runs.md b/internal/manual/chat/adaptive-runs.md index 6837d68522..b16351576c 100644 --- a/internal/manual/chat/adaptive-runs.md +++ b/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 diff --git a/internal/session/team_locality_test.go b/internal/session/team_locality_test.go index a54e221aeb..1341d37ac7 100644 --- a/internal/session/team_locality_test.go +++ b/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 { diff --git a/internal/session/team_relay_test.go b/internal/session/team_relay_test.go index b9cc62be0b..d8d7909578 100644 --- a/internal/session/team_relay_test.go +++ b/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 {