Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions docs/changes/unreleased/2077-full-check-reds.md
Original file line number Diff line number Diff line change
@@ -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.
---
22 changes: 20 additions & 2 deletions internal/relayserve/archived_client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
54 changes: 46 additions & 8 deletions internal/release/promotion_workflow_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,10 @@ import (
"os/exec"
"path/filepath"
"regexp"
"strconv"
"strings"
"testing"
"time"

"gopkg.in/yaml.v3"
)
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand Down
1 change: 1 addition & 0 deletions internal/session/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
7 changes: 7 additions & 0 deletions internal/session/task_index.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
29 changes: 28 additions & 1 deletion internal/session/task_run_belt.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
96 changes: 96 additions & 0 deletions internal/session/task_run_money_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
}
Loading