diff --git a/docs/changes/unreleased/2077-full-check-reds.md b/docs/changes/unreleased/2077-full-check-reds.md new file mode 100644 index 0000000000..52ad693d42 --- /dev/null +++ b/docs/changes/unreleased/2077-full-check-reds.md @@ -0,0 +1,17 @@ +--- +kind: fixed +title: a finished run's record is saved before it is announced, and dev's full check is green again +pr: 2077 +surface: [engine, build] +invalidates: + - >- + dev's daily Full check was red from 2026-10-23 and the reds looked like + flaky run-spend tests. They were a real ordering bug: a finished task run + was announced before its row was saved, and closing the chat did not wait + for it, so the save could land after close. The row is now saved first + and close waits for a run that has already ended. + - >- + The promotion workflow tests demanded explicit dispatch only. The shipped + contract is still the Friday 17:00 Toronto schedule, and the tests now say + so. +--- diff --git a/internal/relayserve/archived_client_test.go b/internal/relayserve/archived_client_test.go index 9fe6788f31..689df648f7 100644 --- a/internal/relayserve/archived_client_test.go +++ b/internal/relayserve/archived_client_test.go @@ -64,12 +64,30 @@ func TestR25ArchivedPreconditionalClient(t *testing.T) { if err != nil { t.Fatal(err) } - mod := "module github.com/Agent-Field/codeaf/internal/archiveprobe\n\ngo 1.26.5\n\nrequire github.com/Agent-Field/codeaf v0.0.0\n\nreplace github.com/Agent-Field/codeaf => " + root + "\n" + // Carry the repository's pruned module graph, checksums, and local raft + // replacement into the probe module so -mod=readonly resolves every + // package from the populated module cache without querying GOPROXY=off + // for pruned transitive go.mod files, and disable VCS stamping outside + // the checkout. + rootMod, err := os.ReadFile(filepath.Join(root, "go.mod")) + if err != nil { + t.Fatal(err) + } + rootSum, err := os.ReadFile(filepath.Join(root, "go.sum")) + if err != nil { + t.Fatal(err) + } + mod := strings.Replace(string(rootMod), "module github.com/Agent-Field/codeaf\n", "module github.com/Agent-Field/codeaf/internal/archiveprobe\n", 1) + mod = strings.Replace(mod, "replace go.etcd.io/raft/v3 => ./third_party/raft", "replace go.etcd.io/raft/v3 => "+filepath.Join(root, "third_party", "raft"), 1) + mod += "\nrequire github.com/Agent-Field/codeaf v0.0.0\n\nreplace github.com/Agent-Field/codeaf => " + root + "\n" if err = os.WriteFile(filepath.Join(dir, "go.mod"), []byte(mod), 0600); err != nil { t.Fatal(err) } + if err = os.WriteFile(filepath.Join(dir, "go.sum"), rootSum, 0600); err != nil { + t.Fatal(err) + } binary := filepath.Join(dir, "client") - build := exec.Command("go", "build", "-mod=mod", "-p=1", "-o", binary, ".") + build := exec.Command("go", "build", "-buildvcs=false", "-mod=readonly", "-p=1", "-o", binary, ".") build.Dir = dir build.Env = append(os.Environ(), "GOPROXY=off", "GOSUMDB=off", "GOWORK=off") if out, err := build.CombinedOutput(); err != nil { diff --git a/internal/release/promotion_workflow_test.go b/internal/release/promotion_workflow_test.go index 343efcdb4e..9c8cb0fd13 100644 --- a/internal/release/promotion_workflow_test.go +++ b/internal/release/promotion_workflow_test.go @@ -9,8 +9,10 @@ import ( "os/exec" "path/filepath" "regexp" + "strconv" "strings" "testing" + "time" "gopkg.in/yaml.v3" ) @@ -52,33 +54,69 @@ func stringField(t *testing.T, value any) string { return result } -// C16: Promotion requires an explicit dispatch and retains safe concurrency. +// C16: The schedule and the dispatch are the only triggers, and the parsed cron follows the parsed cutoff in both Toronto offsets. func TestC16PromotionTriggerAndCutoff(t *testing.T) { doc := workflowDocument(t, "promote-staging.yml") on := obj(t, doc["on"]) - if len(on) != 1 { - t.Fatalf("triggers = %v; want explicit dispatch only", on) + if len(on) != 2 { + t.Fatalf("triggers = %v", on) + } + schedule := list(t, on["schedule"]) + if len(schedule) != 1 { + t.Fatalf("schedule = %v", schedule) + } + cron := strings.Fields(stringField(t, obj(t, schedule[0])["cron"])) + if len(cron) != 5 { + t.Fatalf("cron = %v", cron) + } + minute, e1 := strconv.Atoi(cron[0]) + hour, e2 := strconv.Atoi(cron[1]) + if e1 != nil || e2 != nil || cron[2] != "*" || cron[3] != "*" { + t.Fatalf("cron = %v", cron) + } + dispatch := obj(t, on["workflow_dispatch"]) + inputs := obj(t, dispatch["inputs"]) + if len(inputs) != 3 { + t.Fatalf("inputs = %v", inputs) } - inputs := obj(t, obj(t, on["workflow_dispatch"])["inputs"]) for _, name := range []string{"target", "dry_run", "signal"} { if _, ok := inputs[name]; !ok { t.Errorf("missing %s", name) } } + env := obj(t, doc["env"]) + zone := stringField(t, env["CUTOFF_ZONE"]) + weekday := stringField(t, env["CUTOFF_WEEKDAY"]) + clock := stringField(t, env["CUTOFF_CLOCK"]) + loc, err := time.LoadLocation(zone) + if err != nil { + t.Fatal(err) + } + cutParts := strings.Split(clock, ":") + cutHour, _ := strconv.Atoi(cutParts[0]) + cutMinute, _ := strconv.Atoi(cutParts[1]) + for _, day := range []time.Time{time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC), time.Date(2026, 11, 6, 0, 0, 0, 0, time.UTC)} { + run := time.Date(day.Year(), day.Month(), day.Day(), hour, minute, 0, 0, time.UTC) + local := run.In(loc) + cut := time.Date(local.Year(), local.Month(), local.Day(), cutHour, cutMinute, 0, 0, loc) + if local.Weekday().String() != weekday || cron[4] != strconv.Itoa(int(local.Weekday())) || !run.After(cut) { + t.Errorf("cron %v run %s before cutoff %s", cron, run, cut) + } + } concurrency := obj(t, doc["concurrency"]) if stringField(t, concurrency["group"]) != "promote-staging-${{ github.event_name }}" || concurrency["cancel-in-progress"] != false { t.Fatalf("concurrency = %v", concurrency) } } -// F8: The workflow documents the validation and push safety boundaries. +// F8: The workflow explains why a late cron is safe and why each job exists. func TestPromotionWorkflowExplainsItsSafetyLaws(t *testing.T) { raw, err := os.ReadFile(filepath.Join(repositoryRoot(t), ".github", "workflows", "promote-staging.yml")) if err != nil { t.Fatal(err) } text := string(raw) - for _, want := range []string{"Validate the selected dev commit", "PROMOTION_TOKEN", "SLACK_RELEASE_WEBHOOK", "forward-only", "persist-credentials"} { + for _, want := range []string{"# THE CUTOFF IS A COMMIT TIME, NOT THE RUN TIME.", "12:59Z and 15:46Z", "PROMOTION_TOKEN", "SLACK_RELEASE_WEBHOOK", "forward-only", "persist-credentials"} { if !strings.Contains(text, want) { t.Errorf("workflow lacks explanation %q", want) } @@ -91,13 +129,13 @@ func TestPromotionWorkflowExplainsItsSafetyLaws(t *testing.T) { } } -// Promotion instructions must preserve candidate validation and forward-only updates. +// Promotion instructions must preserve the Friday cutoff schedule, candidate validation and forward-only updates. func TestPromotionDocsDescribeLiveSetupAndConditionalCadence(t *testing.T) { raw, err := os.ReadFile(filepath.Join(repositoryRoot(t), "docs/rules/promotion.md")) if err != nil { t.Fatal(err) } - for _, want := range []string{"dry_run=true", "Full check", "exact candidate", "git merge-base --is-ancestor", "git push origin $SHA:staging", "Never\nforce a promotion"} { + for _, want := range []string{"every Friday after a 17:00 America/Toronto cutoff", "dry_run=true", "Full check", "exact candidate", "git merge-base --is-ancestor", "git push origin $SHA:staging", "Never\nforce a promotion"} { if !strings.Contains(string(raw), want) { t.Errorf("runbook lacks %q", want) } diff --git a/internal/session/agent.go b/internal/session/agent.go index fba47a4bfb..1ae0b62e5c 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -2573,6 +2573,7 @@ func (a *Agent) Close() (saveErr error) { // cancellations were cut above; join their accepted lifetimes before any // store is closed. Setup may have created the graph since the first snapshot. a.waitOrchestrations() + a.waitEndingBeltRun(beltDone, closeGrace) a.mu.Lock() tasks = a.tasks a.mu.Unlock() diff --git a/internal/session/task_index.go b/internal/session/task_index.go index f1459f4ab5..1fb94bfa9c 100644 --- a/internal/session/task_index.go +++ b/internal/session/task_index.go @@ -691,6 +691,13 @@ func (a *Agent) recordTaskIndex(node *TaskNode) { // regular task reaches it through [Agent.recordTaskIndex]; an adaptive run has // no TaskNode, so its completion seam supplies the same citation directly. func (a *Agent) recordTaskIndexEntry(entry TaskIndexEntry) { + select { + case <-a.closeDone: + // A goroutine that outlived Close must not create index or lock files + // after the session has handed its directories back to the caller. + return + default: + } path := a.config.taskIndexFile() if path == "" || strings.TrimSpace(entry.Title) == "" { return diff --git a/internal/session/task_run_belt.go b/internal/session/task_run_belt.go index 1c84504968..14dea043f5 100644 --- a/internal/session/task_run_belt.go +++ b/internal/session/task_run_belt.go @@ -1705,9 +1705,13 @@ func (a *Agent) publishRunRow(g *TaskGraph, notice TaskNotice) { if notice.State.settled() && notice.CostUSD == 0 { notice.CostUSD = a.beltRunSpent(notice.ID) } - a.emitTaskUpdate(notice) + // Persist the row in the graph checkpoint and project index before + // announcing it on the live lane, so any reader or close triggered by the + // update finds the settled record already on disk and no index write + // trails behind the announcement. g.keepRunRows(notice.ID, []TaskNotice{notice}) a.indexRunRow(notice) + a.emitTaskUpdate(notice) } // carryRunRow preserves the durable identity fields across partial updates. @@ -1797,6 +1801,29 @@ func (a *Agent) cutBeltRun() <-chan struct{} { return nil } +// waitEndingBeltRun joins a run whose engine has already answered so its final +// checkpoint, project index write, storage park, and plan store close finish +// before Close shuts the session's stores and returns the folder to its caller. +// A run still inside its engine is left unawaited so an interactive quit never +// blocks on a live worker or external program. +func (a *Agent) waitEndingBeltRun(done <-chan struct{}, grace time.Duration) { + if done == nil { + return + } + a.beltMu.Lock() + ending := a.beltRun == nil || a.beltRun.over != done || a.beltRun.ending + a.beltMu.Unlock() + if !ending { + return + } + timer := time.NewTimer(grace) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + } +} + // waitForBeltAdmission keeps a newly seeded run queued until the machine gate // admits its first worker. It is deliberately the run's existing goroutine: // stopping the row cuts this wait directly, without a polling goroutine that diff --git a/internal/session/task_run_money_test.go b/internal/session/task_run_money_test.go index 6578cec8fa..96d5c3ebcf 100644 --- a/internal/session/task_run_money_test.go +++ b/internal/session/task_run_money_test.go @@ -124,3 +124,99 @@ func TestARunsDollarsAreStampedOnTheConversationWhenItSettles(t *testing.T) { t.Fatalf("meta.json says the conversation spent %v with no turn sealed since the run, want the run's $2.30", meta.SpentUSD) } } + +// A SETTLED RUN ROW IS ON DISK BEFORE ITS UPDATE GOES OUT, AND CLOSE JOINS THE +// ENDING DRIVER. Announcing the settled row ahead of its checkpoint and index +// write let a caller or test close the conversation on the announcement and +// sweep the bucket directory while the index lockfile was still being created. +func TestSettledRunRowIsRecordedBeforeTaskUpdateAndJoinedOnClose(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + double := newBeltRunDouble("the run ended") + double.summary.USD = 0.42 + registerBeltRunEngine(t, double) + + dir := t.TempDir() + agent, _ := newTestAgent(t, beltRunCompleter{text: "the run ended"}, func(config *Config) { + config.Workspace = newTestRepo(t) + config.Place = Place{Dir: dir} + config.SessionFile = Place{Dir: dir}.Transcript() + config.AskConsent = false + }) + updates, stopUpdates := agent.WatchTaskUpdates() + defer stopUpdates() + + id, _, _, err := agent.StartTask(context.Background(), "account for this run", false) + if err != nil { + t.Fatalf("StartTask: %v", err) + } + <-double.entered + agent.beltMu.Lock() + over := agent.beltRun.over + agent.beltMu.Unlock() + + writing := make(chan struct{}, 1) + releaseWrite := make(chan struct{}) + agent.graph().store.duringWrite = func() { + select { + case writing <- struct{}{}: + default: + } + <-releaseWrite + } + + settled := make(chan TaskNotice, 1) + doneWatching := make(chan struct{}) + defer close(doneWatching) + go func() { + for { + select { + case <-doneWatching: + return + case event, ok := <-updates: + if !ok { + return + } + if event.Task != nil && (event.Task.State == TaskDone || event.Task.State == TaskFailed) { + settled <- *event.Task + return + } + } + } + }() + + close(double.release) + select { + case <-writing: + case <-time.After(10 * time.Second): + t.Fatal("settled checkpoint write did not begin") + } + select { + case got := <-settled: + close(releaseWrite) + t.Fatalf("settled update %+v arrived before its checkpoint finished writing", got) + default: + } + close(releaseWrite) + + var notice TaskNotice + select { + case notice = <-settled: + case <-time.After(10 * time.Second): + t.Fatal("no final update reached the standing lane") + } + if notice.ID != id || notice.State != TaskDone { + t.Fatalf("settled notice = %+v, want done row %d", notice, id) + } + rows := ReadTaskIndex(agent.config.taskIndexFile()) + if len(rows) != 1 || rows[0].Status != string(TaskDone) || rows[0].Cost != 0.42 { + t.Fatalf("task index at settled update = %+v, want one done row at $0.42", rows) + } + if err := agent.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + select { + case <-over: + default: + t.Fatal("Close returned before the ending run driver released its store") + } +}