diff --git a/docs/changes/unreleased/1907-stop-owned-work.md b/docs/changes/unreleased/1907-stop-owned-work.md new file mode 100644 index 000000000..4352efd0e --- /dev/null +++ b/docs/changes/unreleased/1907-stop-owned-work.md @@ -0,0 +1,18 @@ +--- +kind: fixed +title: Stop delegated work and helpers at conversation limits +pr: 1907 +surface: [chat, engine] +invalidates: + - "Conversation Stop and time limits previously left delegated runs or helper retries active. Owned work now stops before admission can reopen." + - "History could show a stale record beside the live task, directing users away from its Stop control. The live task now appears once." +--- + +Conversation Stop and `--max-hours` now cancel owned delegated runs, background +jobs and automatic helpers. Stopped runs cannot begin another checking pass or +revive from a pending start. Page Stop also prevents a pending isolated landing +from continuing; a home merge already in progress retains its actual result. +Requests already sent may still report their final usage. + +Task history keeps the live plan row instead of re-adding its saved record, so +opening the task reaches its current room and Stop control. diff --git a/internal/e2e/stop_descendants_wire_e2e_test.go b/internal/e2e/stop_descendants_wire_e2e_test.go new file mode 100644 index 000000000..a52bebf1d --- /dev/null +++ b/internal/e2e/stop_descendants_wire_e2e_test.go @@ -0,0 +1,333 @@ +//go:build e2e + +package e2e + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/config" +) + +// Each paid-looking receipt is local fake data. Request starts, including any +// checker, are counted independently from late accounting for an in-flight call. +func TestTerminalStopEndsOwnedTasksOnTheRealTerminal(t *testing.T) { + if _, err := exec.LookPath("tmux"); err != nil { + t.Skip("tmux required") + } + for _, mode := range []string{"page_stop", "conversation_wall", "unbounded_control"} { + t.Run(mode, func(t *testing.T) { + const model = "stub/terminal-stop" + home := recoveryShortHome(t, newHome(t, map[string]any{"memory.enabled": false, config.KeyModelRoles: "chat:" + model + ",worker:" + model + ",plan:" + model + ",check:" + model, config.KeyTierLowModel: model, config.KeyTierWorkerModel: model, config.KeyTierHighModel: model})) + ws := newWorkspace(t, "terminal-stop", false) + var mu sync.Mutex + starts := 0 + issued := map[string]bool{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/models": + _ = json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{"id": model, "context_length": 131072, "pricing": map[string]string{"prompt": "0.000001", "completion": "0.000001"}, "supported_parameters": []string{"tools", "tool_choice"}}}}) + case "/key", "/auth/key", "/credits": + fmt.Fprint(w, `{"data":{"limit_remaining":100,"total_credits":100,"total_usage":0}}`) + case "/chat/completions": + var body struct { + Stream bool `json:"stream"` + Messages []struct { + Content json.RawMessage `json:"content"` + } `json:"messages"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "fixture decode", 400) + return + } + var assignment []string + for _, msg := range body.Messages { + if match := freeConsentAssignment.FindStringSubmatch(string(msg.Content)); len(match) == 3 { + assignment = match + break + } + } + if len(assignment) != 3 { + terminalStopResponse(w, model, "Acknowledged.", nil, body.Stream) + return + } + mu.Lock() + starts++ + first := !issued[assignment[1]] + issued[assignment[1]] = true + mu.Unlock() + if first { + command := "i=0; while [ $i -lt 120 ]; do printf 'tick\\n' >> stop-heartbeat.txt; i=$((i+1)); sleep 1; done" + args, _ := json.Marshal(map[string]string{"command": command}) + terminalStopResponse(w, model, "", map[string]any{"index": 0, "id": "stop-loop-" + assignment[2], "type": "function", "function": map[string]string{"name": "bash", "arguments": string(args)}}, body.Stream) + } else { + terminalStopResponse(w, model, "The long task is still in progress.", nil, body.Stream) + } + default: + http.NotFound(w, r) + } + })) + defer server.Close() + env := []string{config.APIKeyEnv + "=fixture-terminal-stop-key", "CODEAF_BASE_URL=" + server.URL, "CODEAF_PROFILE_DIR="} + args := []string{"chat", "--model", model, "--one-model", "--no-host", "--yolo"} + if mode == "conversation_wall" { + args = append(args, "--max-hours", "0.0044444444") + } + rig := startWithEnv(t, env, "terminal_"+mode, home, ws, 170, 48, args...) + rig.skipSetup(t) + rig.lit("/task solo run a slow counting loop for two minutes, writing each tick to stop-heartbeat.txt using bash; do not finish early") + rig.keys("Enter") + deadline := time.Now().Add(12 * time.Second) + for stopHeartbeatBytes(home) == 0 && stopHeartbeatBytes(ws) == 0 { + if time.Now().After(deadline) { + t.Fatalf("owned task never started its actual loop: %s", rig.capture()) + } + time.Sleep(pollEvery) + } + if mode == "unbounded_control" { + before := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws) + time.Sleep(18 * time.Second) + if after := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws); after <= before { + t.Fatalf("unbounded positive control stopped early: before=%d after=%d", before, after) + } + if strings.Contains(rig.capture(), "conversation time limit reached") { + t.Fatal("unbounded control acquired a time limit") + } + } + if mode != "conversation_wall" { + stopOwnedTaskThroughItsPage(t, rig) + } + if mode == "conversation_wall" { + rig.waitFor(35*time.Second, "conversation time limit reached") + if screen := rig.capture(); strings.Contains(screen, "background work") && strings.Contains(screen, "ctrl+e") { + rig.keys("C-e") + } + screen := rig.waitFor(10*time.Second, "· stopped ·") + if !strings.Contains(screen, "· stopped ·") { + t.Fatal("joined wall stop summary was not disclosed") + } + if !stopRecordedReason(home, "a time limit you set stopped it") { + t.Fatal("joined wall stop result did not record its exact reason") + } + } else { + rig.waitFor(20*time.Second, "stopped") + } + // Completion acknowledgment is the boundary: do not conflate a receipt from + // an already-dispatched call with permission to start a checker or resume. + mu.Lock() + atStop := starts + mu.Unlock() + before := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws) + spendAtStop := stopSpendSnapshot(t, home) + if spendAtStop.Cost <= 0 || spendAtStop.Cached == 0 { + t.Fatalf("fixture failed to meter nonzero spend/cache: %+v", spendAtStop) + } + for end := time.Now().Add(10 * time.Second); time.Now().Before(end); { + screen := rig.capture() + if strings.Contains(screen, "a accept n not right s tell it") || strings.Contains(screen, "Working checking") { + t.Fatalf("stopped task entered acceptance/checking: %s", screen) + } + mu.Lock() + now := starts + mu.Unlock() + if now != atStop { + t.Fatalf("new task/checker model requests after terminal stop: %d -> %d", atStop, now) + } + time.Sleep(200 * time.Millisecond) + } + if afterSpend := stopSpendSnapshot(t, home); afterSpend != spendAtStop { + t.Fatalf("tokens/cache/spend changed after joined stop: before=%+v after=%+v", spendAtStop, afterSpend) + } + after := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws) + if after != before { + t.Fatalf("owned child kept writing after stop: bytes %d -> %d", before, after) + } + t.Logf("PASS real terminal %s: owned child stopped, model request starts stable=%d for10s, heartbeat bytes stable=%d; no checking/acceptance/resume.\n%s", mode, atStop, after, rig.capture()) + }) + } +} + +func stopHeartbeatBytes(root string) int64 { + var size int64 + _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == "stop-heartbeat.txt" { + if info, e := d.Info(); e == nil { + size += info.Size() + } + } + return nil + }) + return size +} + +func TestTerminalStopLiveOwnedTaskWorkflow(t *testing.T) { + requireTmuxAndKey(t) + const model = "deepseek/deepseek-v4.1-flash" + if e2eModel != model { + t.Fatalf("requires CODEAF_E2E_MODEL=%s", model) + } + home := recoveryShortHome(t, newHome(t, map[string]any{"memory.enabled": false})) + ws := newWorkspace(t, "live-terminal-stop", false) + env := []string{config.APIKeyEnv + "=" + liveKey(t), "CODEAF_BASE_URL=https://openrouter.ai/api/v1", "CODEAF_PROFILE_DIR="} + rig := startWithEnv(t, env, "terminal_stop_live", home, ws, 170, 48, "chat", "--model", model, "--one-model", "--no-host", "--yolo") + rig.skipSetup(t) + prompt := "We are testing our fictional counting workflow. Acknowledge that a long-running counter should retain its partial output when I explicitly stop it." + rig.lit(prompt) + rig.keys("Enter") + if answer, _ := privateMemoryAwaitAnswer(home, prompt, 90*time.Second); answer == "" { + t.Fatal("live introduction did not complete") + } + rig.lit("/task solo use bash to run a counting loop for120 seconds, appending one line tick per second to stop-heartbeat.txt in your working directory. Run the loop now, wait for it, and do not mark this task done before the loop finishes.") + rig.keys("Enter") + deadline := time.Now().Add(2 * time.Minute) + for stopHeartbeatBytes(home)+stopHeartbeatBytes(ws) == 0 { + if time.Now().After(deadline) { + t.Fatalf("live task never started its counter: %s", rig.capture()) + } + time.Sleep(pollEvery) + } + stopOwnedTaskThroughItsPage(t, rig) + rig.waitFor(20*time.Second, "stopped") + bytesAtStop := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws) + time.Sleep(10 * time.Second) + if got := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws); got != bytesAtStop { + t.Fatalf("live stopped child kept writing: %d -> %d", bytesAtStop, got) + } + screen := rig.capture() + if strings.Contains(screen, "Working checking") || strings.Contains(screen, "a accept n not right s tell it") { + t.Fatalf("live stopped task entered checking or acceptance: %s", screen) + } + models := callModels(t, home) + if len(models) == 0 { + t.Fatal("no live receipts") + } + for _, got := range models { + if got != model { + t.Fatalf("unexpected live model %s", got) + } + } + t.Logf("PASS live task: actual long-running bash counter stopped through task page; partial output retained and unchanged10s; no acceptance/checking; all %d recorded calls exact %s.\n%s", len(models), model, screen) +} + +func terminalStopResponse(w http.ResponseWriter, model, text string, call any, stream bool) { + message := map[string]any{"role": "assistant", "content": text} + finish := "stop" + if call != nil { + message["tool_calls"] = []any{call} + finish = "tool_calls" + } + usage := map[string]any{"prompt_tokens": 40, "completion_tokens": 10, "total_tokens": 50, "cost": 0.02, "prompt_tokens_details": map[string]int{"cached_tokens": 10}} + if stream { + w.Header().Set("Content-Type", "text/event-stream") + for _, choice := range []map[string]any{{"index": 0, "delta": message, "finish_reason": nil}, {"index": 0, "delta": map[string]any{}, "finish_reason": finish}} { + raw, _ := json.Marshal(map[string]any{"id": "terminal-stop-fixture", "model": model, "choices": []any{choice}, "usage": usage}) + fmt.Fprintf(w, "data: %s\n\n", raw) + } + fmt.Fprint(w, "data: [DONE]\n\n") + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"id": "terminal-stop-fixture", "model": model, "choices": []any{map[string]any{"index": 0, "message": message, "finish_reason": finish}}, "usage": usage}) +} + +type stopSpend struct { + Prompt, Completion, Cached int + Cost float64 +} + +func stopSpendSnapshot(t *testing.T, home string) stopSpend { + t.Helper() + raw, err := os.ReadFile(filepath.Join(home, "logs", "calls.jsonl")) + if err != nil { + t.Fatal(err) + } + var total stopSpend + for _, line := range strings.Split(string(raw), "\n") { + var row struct { + Prompt int `json:"prompt_tokens"` + Completion int `json:"completion_tokens"` + Cached int `json:"cached_tokens"` + Cost float64 `json:"cost"` + } + if json.Unmarshal([]byte(line), &row) == nil { + total.Prompt += row.Prompt + total.Completion += row.Completion + total.Cached += row.Cached + total.Cost += row.Cost + } + } + return total +} + +func stopOwnedTaskThroughItsPage(t *testing.T, rig *rig) { + t.Helper() + openTasksPlace(t, rig) + if screen := rig.capture(); !strings.Contains(screen, "1 chat · 1 subtask") { + t.Fatalf("fresh single-task history duplicated its active run: %s", screen) + } + // Select the live plan by its advertised Stop verb, not by a title or + // row index. The single-task assertion above also catches duplicate records. + var screen string + selected := false + for attempts := 0; attempts < 8; attempts++ { + screen = rig.capture() + if strings.Contains(screen, "s stop it") { + selected = true + break + } + rig.keys("Down") + } + if !selected { + t.Fatalf("no active plan row offered Stop: %s", screen) + } + t.Logf("selected stoppable task row before enter:\n%s", screen) + rig.keys("Enter") + screen = rig.waitFor(15*time.Second, say(t, "roomTabsWords")) + if !strings.Contains(screen, say(t, "roomTabsWords")) { + t.Fatal("active task room did not load") + } + rig.keys("x") + screen = rig.waitFor(10*time.Second, "Stop this task?", "stop it", "keep going") + if !strings.Contains(screen, "Stop this task?") || !strings.Contains(screen, "stop it") || !strings.Contains(screen, "keep going") { + t.Fatal("actual Stop confirmation was not displayed") + } + rig.keys("1") + rig.keys("Enter") +} + +func stopRecordedReason(home, reason string) bool { + found := false + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err != nil || d.IsDir() || d.Name() != "tasks.json" { + return nil + } + raw, e := os.ReadFile(path) + if e != nil { + return nil + } + var record struct { + Runs []struct { + Stopped bool `json:"stopped"` + Report string `json:"report"` + } `json:"runs"` + } + if json.Unmarshal(raw, &record) == nil { + for _, run := range record.Runs { + if run.Stopped && strings.Contains(run.Report, reason) { + found = true + } + } + } + return nil + }) + return found +} diff --git a/internal/manual/chat/running-from-the-terminal.md b/internal/manual/chat/running-from-the-terminal.md index 2c2b398e2..d74574744 100644 --- a/internal/manual/chat/running-from-the-terminal.md +++ b/internal/manual/chat/running-from-the-terminal.md @@ -404,6 +404,15 @@ The top-level synopsis includes `do` worker limits and model choices (`--slots`, `--best`, `--cheap`, `--pin`, and `--check-model`). It also names chat `--max-cost` and `--max-hours`; those unattended-work limits require `--yolo`. +## Conversation time limit — --max-hours + +When the conversation reaches `--max-hours`, its delegated runs, background jobs, +and automatic memory or title work stop before the terminal error appears: +`conversation time limit reached · start a new conversation or relaunch with a larger --max-hours`. +The stopped run records `a time limit you set stopped it`. Start a new conversation +or relaunch with a larger limit to do more work. Closing a window is different: it +detaches the window and does not itself stop delegated work. + ## What $? means after a headless one-shot — the codes it leaves with **One table, and `codeaf do`, `codeaf exec` and `codeaf run` all leave on it.** diff --git a/internal/manual/chat/task-controls.md b/internal/manual/chat/task-controls.md index 1bff6947b..71fc8345a 100644 --- a/internal/manual/chat/task-controls.md +++ b/internal/manual/chat/task-controls.md @@ -69,7 +69,8 @@ stop confirmations. The header and controls stop the work on the open page. conversation and the model calls `tasks` with that id and `stop`. That is the same door `/stop`, `x` and the `Stop task…` row take, so the ending is identical whichever hand pulls it: the worker is cut off where it stands, its branch is kept with its work committed on it, -the row reads `stopped`, and what it spent freezes where it was. +the row reads `stopped`, and no new requests are started for it. Usage from a request +already sent can still arrive in its final receipt. **A hosted conversation waits for the stop to come back before it claims success.** Once the normal session host has ended the run and returned its receipt, the chat can answer `Stopped.`. diff --git a/internal/manual/chat/worker-harness.md b/internal/manual/chat/worker-harness.md index d20e5c891..250a282ae 100644 --- a/internal/manual/chat/worker-harness.md +++ b/internal/manual/chat/worker-harness.md @@ -646,6 +646,12 @@ task no `p pause` is named. Closing the window, `ctrl+c` and `/quit` do NOT stop a run: it carries on without the window and its work lands by itself. Stop it first if you want it ended. +Stopping a delegated run also stops its workers and prevents another checking or +acceptance pass from starting. Its saved work and trace remain available. A request +already sent can still return its final usage receipt; stopping does not erase costs +already incurred. If bringing finished work home has already begun, that operation +finishes and its actual result is preserved. + ## What a stopped run keeps: its branch after you stop it, bring it in or drop it **Nothing a stopped run made goes into your folder.** What it had made so far is diff --git a/internal/session/conversation_wall.go b/internal/session/conversation_wall.go new file mode 100644 index 000000000..9c004d158 --- /dev/null +++ b/internal/session/conversation_wall.go @@ -0,0 +1,64 @@ +package session + +import "time" + +const conversationTimeLimitReason = "a time limit you set stopped it" + +// The launch wall owns the conversation's work, not the attached window. A +// window can detach without ending a run; expiry is an explicit terminal limit. +func (a *Agent) armConversationWall() { + a.mu.Lock() + if a.closed || a.wallStop != nil { + a.mu.Unlock() + return + } + stop := make(chan struct{}) + a.wallStop = stop + left := time.Until(a.startedAt.Add(a.config.Budget.Wall)) + a.mu.Unlock() + go func() { + timer := time.NewTimer(left) + defer timer.Stop() + select { + case <-stop: + return + case <-timer.C: + } + a.endConversationOnWall() + }() +} + +func (a *Agent) endConversationOnWall() { + a.mu.Lock() + if a.closed || a.wallEndingTaken { + a.mu.Unlock() + return + } + a.wallEndingTaken = true + a.mu.Unlock() + if err := a.stopWorkFor(conversationTimeLimitReason, StopByTimeLimit); err != nil { + return + } + a.mu.Lock() + done := a.workStopDone + a.mu.Unlock() + if done != nil { + <-done + } + a.mu.Lock() + defer a.mu.Unlock() + if a.closed { + return + } + // The terminal error is published only after owned producers have joined. + for _, lane := range a.wakeLanes { + stream := newEventStream() + select { + case lane <- stream.out: + stream.send(Event{Kind: EventError, Err: launchBudgetReached{said: conversationTimeLimitWord}}) + stream.close() + default: + stream.close() + } + } +} diff --git a/internal/session/handlepick.go b/internal/session/handlepick.go index 5d5ff3daa..761880daa 100644 --- a/internal/session/handlepick.go +++ b/internal/session/handlepick.go @@ -88,6 +88,12 @@ func (a *Agent) chooseTeamHandlesLater(title string, roles []teamRole) { if !rolesWantHandle(roles) { return } + a.mu.Lock() + stopped := a.closed || a.workStopped + a.mu.Unlock() + if stopped { + return + } a.team.mu.Lock() if a.team.handleTried { a.team.mu.Unlock() @@ -97,7 +103,7 @@ func (a *Agent) chooseTeamHandlesLater(title string, roles []teamRole) { a.team.mu.Unlock() a.mu.Lock() ctx, model, closed := a.titleCtx, a.model, a.closed - if ctx == nil || closed { + if ctx == nil || closed || a.workStopped { a.mu.Unlock() return } diff --git a/internal/session/interactive_budget.go b/internal/session/interactive_budget.go index c4a805a30..909831c77 100644 --- a/internal/session/interactive_budget.go +++ b/internal/session/interactive_budget.go @@ -11,6 +11,8 @@ import ( // person as principal must not discard a limit supplied at the door. var ErrLaunchBudget = errors.New("session: launch limit reached") +const conversationTimeLimitWord = "conversation time limit reached · start a new conversation or relaunch with a larger --max-hours" + func (a *Agent) interactiveBudget() Budget { if !a.config.Interactive || a.config.InTask || a.config.Errand { return Budget{} @@ -19,12 +21,12 @@ func (a *Agent) interactiveBudget() Budget { } // launchBudgetBlockLocked runs at the existing turn-admission boundary with -// a.mu held. Like the conversation spend limit, it lets in-flight work finish; -// it is not a timer that interrupts tools or an exact reservation across tasks. +// a.mu held. This check does not reserve spending across tasks. The independent +// conversation wall reader cancels owned work when the launch time expires. func (a *Agent) launchBudgetBlockLocked() error { budget := a.interactiveBudget() if budget.Wall > 0 && !a.startedAt.IsZero() && time.Since(a.startedAt) >= budget.Wall { - return launchBudgetReached{said: "conversation time limit reached · start a new conversation or relaunch with a larger --max-hours"} + return launchBudgetReached{said: conversationTimeLimitWord} } if budget.USD > 0 && a.usage.CostUSD >= budget.USD { return launchBudgetReached{said: fmt.Sprintf("conversation launch limit reached · %s spent of %s · relaunch with a larger --max-cost", diff --git a/internal/session/memory.go b/internal/session/memory.go index feb70480b..81e62e24a 100644 --- a/internal/session/memory.go +++ b/internal/session/memory.go @@ -1272,7 +1272,7 @@ func appendMemoryFailureLog(path, via, reason, storeErr string) { func (a *Agent) startMemoryJob() bool { a.mu.Lock() defer a.mu.Unlock() - if a.closed || a.memoryCtx == nil { + if a.closed || a.workStopped || a.memoryCtx == nil { return false } a.memoryJobs.Add(1) diff --git a/internal/session/session.go b/internal/session/session.go index 439bacef1..d86917a60 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -3014,7 +3014,7 @@ type Agent struct { // read off the background. wokenTurnBase func() context.Context // wallStop ends the one reader that waits independently for this session's - // wall, and is nil when the session has no unattended wall (wallclock.go). + // wall, and is nil when neither launch nor Steward sets a wall. // wallEndingTaken is the once-only mark on the ending that reader writes. // Both are guarded by mu because Close and the reader meet on them from // different goroutines. @@ -3054,8 +3054,11 @@ type Agent struct { // never has to, because a cut is refused while a turn is in flight. turnFloor int // Explicit conversation stops suppress autonomous wakes until fresh input. - workStopped bool - workStopping bool + workStopped bool + workStopping bool + workEpoch uint64 + workStopReason string + workStopDone chan struct{} // cancel ends the turn in flight AND SAYS WHICH DOOR IT CAME THROUGH. It is // a [context.CancelCauseFunc] rather than a plain one because a turn that // ends with nothing said has to be able to account for itself afterwards — diff --git a/internal/session/stop_auxiliary_test.go b/internal/session/stop_auxiliary_test.go new file mode 100644 index 000000000..6c9b2ac58 --- /dev/null +++ b/internal/session/stop_auxiliary_test.go @@ -0,0 +1,79 @@ +package session + +import ( + "context" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/Agent-Field/agentfield/sdk/go/ai" +) + +type stoppedMemoryRetryProbe struct { + calls atomic.Int64 + first chan struct{} +} + +func (p *stoppedMemoryRetryProbe) CompleteWithMessages(context.Context, []ai.Message, ...ai.Option) (*ai.Response, error) { + if p.calls.Add(1) == 1 { + close(p.first) + return nil, context.DeadlineExceeded + } + return textResponse(`{"keep":""}`), nil +} + +func TestConversationStopPreventsBackgroundMemoryRetryAndRestoresOnlyForFreshInput(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + memory, stopMemory := context.WithCancel(context.Background()) + title, stopTitle := context.WithCancel(context.Background()) + t.Cleanup(stopMemory) + t.Cleanup(stopTitle) + agent := &Agent{memoryCtx: memory, memoryStop: stopMemory, titleCtx: title, titleStop: stopTitle} + oldMemory, oldTitle := agent.memoryCtx, agent.titleCtx + probe := &stoppedMemoryRetryProbe{first: make(chan struct{})} + if !agent.startMemoryJob() { + t.Fatal("memory job not admitted") + } + go func() { + defer agent.memoryJobs.Done() + _, _ = agent.noticeWithRecovery(probe, "mock preference", "okay", nil, true) + }() + <-probe.first + if err := agent.StopWork(); err != nil { + t.Fatal(err) + } + time.Sleep(3 * time.Second) + if got := probe.calls.Load(); got != 1 { + t.Fatalf("memory started %d model calls after Stop; want first call only", got) + } + if oldMemory.Err() == nil || oldTitle.Err() == nil { + t.Fatal("stopped auxiliary lifetimes remain able to retry") + } + if agent.startMemoryJob() { + agent.memoryJobs.Done() + t.Fatal("new memory job admitted while stopped") + } + agent.mu.Lock() + if err := agent.resumeWorkLocked(); err != nil { + agent.mu.Unlock() + t.Fatal(err) + } + resumedMemory, resumedTitle := agent.memoryCtx, agent.titleCtx + agent.mu.Unlock() + if resumedMemory == oldMemory || resumedTitle == oldTitle || resumedMemory.Err() != nil || resumedTitle.Err() != nil { + t.Fatal("fresh input did not get fresh auxiliary lifetimes") + } + }) +} + +func TestConversationStopRejectsLateTeamHandleAttempt(t *testing.T) { + agent := &Agent{workStopped: true, config: Config{ProfileDir: t.TempDir(), SessionFile: "mock-session.jsonl"}} + agent.chooseTeamHandlesLater("Mock project", []teamRole{{derived: true}}) + agent.team.mu.Lock() + attempted := agent.team.handleTried + agent.team.mu.Unlock() + if attempted { + t.Fatal("stopped conversation admitted a new auxiliary handle attempt") + } +} diff --git a/internal/session/stop_descendants_test.go b/internal/session/stop_descendants_test.go new file mode 100644 index 000000000..286ae1cf5 --- /dev/null +++ b/internal/session/stop_descendants_test.go @@ -0,0 +1,113 @@ +package session + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +func TestConversationStopCutsItsOwnedDelegatedRun(t *testing.T) { + agent, engine, _, _ := stoppableBeltRun(t, 91) + if err := agent.StopWork(); err != nil { + t.Fatal(err) + } + deadline := time.After(250 * time.Millisecond) + select { + case <-engine.ctx.Done(): + case <-deadline: + t.Fatal("conversation Stop returned but its delegated run could still spend") + } +} + +func TestInteractiveConversationWallCutsItsOwnedDelegatedRun(t *testing.T) { + agent, engine, _, _ := stoppableBeltRun(t, 92) + lane, unwatch := agent.WatchWakes() + defer unwatch() + agent.mu.Lock() + agent.config.Interactive = true + agent.config.Budget = Budget{Wall: 20 * time.Millisecond} + agent.startedAt = time.Now() + agent.mu.Unlock() + agent.armWallClock() + select { + case <-engine.ctx.Done(): + case <-time.After(250 * time.Millisecond): + t.Fatal("conversation time limit left its delegated run able to spend") + } + stream := awaitWallStream(t, lane, time.Second) + var terminal bool + for event := range stream { + if event.Kind == EventError && event.Err != nil && event.Err.Error() == conversationTimeLimitWord { + terminal = true + } + } + if !terminal { + t.Fatal("time limit did not publish its actionable terminal error") + } + agent.beltMu.Lock() + live := agent.beltRun != nil + agent.beltMu.Unlock() + if live { + t.Fatal("terminal time-limit error arrived before owned run joined") + } + agent.mu.Lock() + err := agent.resumeWorkLocked() + agent.mu.Unlock() + if err == nil || err.Error() != conversationTimeLimitWord { + t.Fatalf("fresh input revived an expired conversation: %v", err) + } +} + +type stoppingLandingEngine struct { + *beltRunDouble + landingEntered chan struct{} + releaseLanding chan struct{} +} + +func (e *stoppingLandingEngine) Land(context.Context, *plandb.Store, string, string, string) (RunLanding, error) { + close(e.landingEntered) + <-e.releaseLanding + return e.landing, nil +} + +func TestPageStopDuringLandingRemainsTerminal(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + workspace := beltRunCommittedRepo(t) + engine := &stoppingLandingEngine{beltRunDouble: newBeltRunDouble("work done"), landingEntered: make(chan struct{}), releaseLanding: make(chan struct{})} + registerBeltRunEngine(t, engine) + agent, _ := newTestAgent(t, beltRunCompleter{text: "done"}, func(c *Config) { c.Workspace = workspace; c.Place = Place{Dir: t.TempDir()} }) + if err := agent.startKnownTaskRun(context.Background(), 93, "fixture work", "brief", nil, taskStand{dir: workspace, mode: TaskModeWorktree}, ""); err != nil { + t.Fatal(err) + } + <-engine.entered + close(engine.release) + <-engine.landingEntered + if err := agent.PlanCancel("93"); err != nil { + t.Fatal(err) + } + close(engine.releaseLanding) + beltRunWaitFor(t, "stopped landing to settle", func() bool { agent.beltMu.Lock(); defer agent.beltMu.Unlock(); return agent.beltRun == nil }) + for _, row := range agent.graph().runRows(93) { + if row.ID == 93 && (!row.Stopped || !strings.Contains(row.Report, "stopped")) { + t.Fatalf("page Stop was replaced by a normal landing: %+v", row) + } + } +} + +func TestLegacyWallStopsAStandaloneJobWithoutDelegatedTasks(t *testing.T) { + agent, _ := newTestAgent(t, &scriptedCompleter{}, func(c *Config) { + c.Unattended = true + c.Budget = Budget{Wall: time.Hour} + }) + id := startJob(t, agent, "sleep 30") + spendWall(t, agent) + if !agent.endRunOnTheWall() { + t.Fatal("expired Steward wall ignored a standalone background job") + } + if agent.jobs.find(id).running() { + t.Fatal("wall ending arrived before the background job joined") + } +} diff --git a/internal/session/stopcause.go b/internal/session/stopcause.go index a9e1f365b..93f4129cc 100644 --- a/internal/session/stopcause.go +++ b/internal/session/stopcause.go @@ -59,6 +59,8 @@ const ( // StopByWorkStopped is [Agent.StopWork]: everything this conversation had // running was stopped at once (stopwork.go). StopByWorkStopped StopDoor = "work stopped" + // StopByTimeLimit is the conversation wall ending its owned work. + StopByTimeLimit StopDoor = "conversation time limit" // StopByRetired is a hosted session let go of for want of anybody attached // to it (internal/remote). StopByRetired StopDoor = "session retired" @@ -167,6 +169,8 @@ func stopSentence(door StopDoor, name string) string { return "the reply stopped when this conversation was left — ask again to pick it up" case StopByAbandoned: return "the reply was let go of after the stop took too long" + case StopByTimeLimit: + return "a time limit you set stopped it" case StopByWorkStopped: return "everything running here was stopped, the reply with it" case StopByRetired: diff --git a/internal/session/stoprun.go b/internal/session/stoprun.go index b78455d68..f98279d75 100644 --- a/internal/session/stoprun.go +++ b/internal/session/stoprun.go @@ -71,21 +71,48 @@ func beltStoppedWhere(branch, ground string, changed []string) string { // every other task number lives. func (a *Agent) stopBeltRow(id uint64, why string) (string, bool, error) { why = strings.TrimSpace(why) - a.beltMu.Lock() - run := a.beltRun - if run == nil { + for { + a.beltMu.Lock() + run := a.beltRun a.beltMu.Unlock() - return a.endedBeltRow(id, why) + if run == nil { + return a.endedBeltRow(id, why) + } + // Never acquire delivery admission while holding beltMu: acceptance takes + // the conversation lock and may read this run's state. + run.deliveryMu.Lock() + a.beltMu.Lock() + if a.beltRun != run { + a.beltMu.Unlock() + run.deliveryMu.Unlock() + continue + } + return a.stopCurrentBeltRow(run, id, why) } +} + +// stopCurrentBeltRow enters holding deliveryMu and beltMu. Both are released +// with the stop flag, before cancellation, store writes or other work. +func (a *Agent) stopCurrentBeltRow(run *beltRun, id uint64, why string) (string, bool, error) { + unlock := func() { a.beltMu.Unlock(); run.deliveryMu.Unlock() } if run.row == id { name := taskStopName(id, run.title) + if run.bringingHome { + run.quiet = true + cut := run.cut + unlock() + if cut != nil { + cut() + } + return name + " has already finished; there is nothing to stop", true, nil + } if run.stopped { - a.beltMu.Unlock() + unlock() return name + " is already stopping", true, nil } run.stopped, run.stopReason = true, why cut := run.cut - a.beltMu.Unlock() + unlock() // THE STORE FIRST, THE CONTEXT NEXT (the header says why). A store that // would not take the write is said on the run's page by whoever settles // it; the context is cut either way, because the person asked for the @@ -108,7 +135,7 @@ func (a *Agent) stopBeltRow(id uint64, why string) (string, bool, error) { for _, row := range run.joined { joined = joined || row == id } - a.beltMu.Unlock() + unlock() if !joined { return a.endedBeltRow(id, why) } diff --git a/internal/session/stopwork.go b/internal/session/stopwork.go index 8b3958ab1..625271863 100644 --- a/internal/session/stopwork.go +++ b/internal/session/stopwork.go @@ -1,13 +1,18 @@ package session -import "errors" +import ( + "context" + "errors" +) var errWorkStopping = errors.New("this conversation is stopping; wait for its work to finish stopping before sending a new message") // StopWork closes this conversation's admission gates before cancelling work. // Unlike Interrupt, this includes background jobs and adaptive runs. Completion // news remains in the record but cannot buy another turn until a fresh Submit. -func (a *Agent) StopWork() error { +func (a *Agent) StopWork() error { return a.stopWorkFor("", StopByWorkStopped) } + +func (a *Agent) stopWorkFor(why string, door StopDoor) error { a.mu.Lock() if a.closed { a.mu.Unlock() @@ -18,9 +23,13 @@ func (a *Agent) StopWork() error { return nil } a.workStopped, a.workStopping = true, true + a.workEpoch++ + a.workStopReason = why + a.workStopDone = make(chan struct{}) a.dropFollowUpsLocked() a.stopSteerGraceLocked() cancel, done := a.cancel, a.done + memoryStop, titleStop := a.memoryStop, a.titleStop // Keep completed runs inspectable, and mark cancellation as the person's // decision before cutting their contexts. Close deliberately discards this // registry; stopping a conversation must preserve it. @@ -46,9 +55,15 @@ func (a *Agent) StopWork() error { graph.mu.Unlock() } a.mu.Unlock() + if memoryStop != nil { + memoryStop() + } + if titleStop != nil { + titleStop() + } a.interrupt.begin() if cancel != nil { - cancel(stopFor(StopByWorkStopped)) + cancel(stopFor(door)) } for _, live := range runs { live.run.Cancel() @@ -59,17 +74,18 @@ func (a *Agent) StopWork() error { } // The UI receives acceptance promptly. Reopening admission waits for the old // producers themselves, so a late tool cannot join a freshly resumed registry. - go a.finishWorkStop(done, graph, jobs) + beltDone := a.stopOwnedBeltRun(why) + go a.finishWorkStop(done, graph, jobs, beltDone, why) return nil } -func (a *Agent) finishWorkStop(done <-chan struct{}, graph *TaskGraph, jobs *jobRegistry) { +func (a *Agent) finishWorkStop(done <-chan struct{}, graph *TaskGraph, jobs *jobRegistry, beltDone <-chan struct{}, why string) { if graph != nil { graph.mu.Lock() ids := append([]uint64(nil), graph.order...) graph.mu.Unlock() for _, id := range ids { - _, _ = graph.stop(id) + _, _ = graph.stopFor(id, why) } } if jobs != nil { @@ -78,6 +94,11 @@ func (a *Agent) finishWorkStop(done <-chan struct{}, graph *TaskGraph, jobs *job if done != nil { <-done } + if beltDone != nil { + <-beltDone + } + a.memoryJobs.Wait() + a.titleJobs.Wait() a.orchestrateWorkers.Wait() // Setup admitted before StopWork may have created the graph meanwhile. Its // constructor inherits the closed gate, and its final nodes are cancelled here. @@ -89,7 +110,7 @@ func (a *Agent) finishWorkStop(done <-chan struct{}, graph *TaskGraph, jobs *job ids := append([]uint64(nil), graph.order...) graph.mu.Unlock() for _, id := range ids { - _, _ = graph.stop(id) + _, _ = graph.stopFor(id, why) } graph.runners.Wait() } @@ -98,20 +119,40 @@ func (a *Agent) finishWorkStop(done <-chan struct{}, graph *TaskGraph, jobs *job <-job.done } } + // A run admitted before Stop may have finished setup after the first cut. + a.beltStartMu.Lock() + late := a.stopOwnedBeltRun(why) + a.beltStartMu.Unlock() + if late != nil { + <-late + } a.mu.Lock() a.workStopping = false + close(a.workStopDone) a.mu.Unlock() } // resumeWorkLocked is reached only by a new explicit user submission. Keeping // the same registries preserves job IDs, logs and the task history on reopening. func (a *Agent) resumeWorkLocked() error { + if a.wallEndingTaken && a.interactiveBudget().Wall > 0 { + return launchBudgetReached{said: conversationTimeLimitWord} + } if a.workStopping { return errWorkStopping } if !a.workStopped { return nil } + // The old auxiliary producers have joined before admission reopens. A + // fresh explicit input gets fresh lifetimes; old retries keep cancelled ones. + if a.memoryCtx != nil { + a.memoryCtx, a.memoryStop = context.WithCancel(context.Background()) + } + a.titleCtx, a.titleStop = context.WithCancel(context.Background()) + if a.title == "" { + a.titleTried = false + } a.workStopped = false a.orchestrateContext, a.orchestrateStop = nil, nil if a.jobs != nil { @@ -124,3 +165,16 @@ func (a *Agent) resumeWorkLocked() error { } return nil } + +// stopOwnedBeltRun marks the run terminal before cutting its shared worker +// context. Close uses a different road so ordinary detach stays resumable. +func (a *Agent) stopOwnedBeltRun(why string) <-chan struct{} { + a.beltMu.Lock() + run := a.beltRun + a.beltMu.Unlock() + if run == nil { + return nil + } + _, _, _ = a.stopBeltRow(run.row, why) + return run.over +} diff --git a/internal/session/task_run_admission.go b/internal/session/task_run_admission.go new file mode 100644 index 000000000..c30b8c73f --- /dev/null +++ b/internal/session/task_run_admission.go @@ -0,0 +1,54 @@ +package session + +import "github.com/Agent-Field/codeaf/internal/router" + +// A request waiting for the start lock belongs to the admission epoch in which +// it arrived. A later explicit Submit cannot revive a request cut by Stop. +func (a *Agent) beltAdmissionEpoch() (uint64, error) { + a.mu.Lock() + defer a.mu.Unlock() + if a.closed { + return 0, errAgentClosed + } + if a.workStopped { + return 0, errWorkStopping + } + return a.workEpoch, nil +} +func (a *Agent) checkBeltAdmission(epoch uint64) error { + a.mu.Lock() + defer a.mu.Unlock() + if a.closed { + return errAgentClosed + } + if a.workStopped || a.workEpoch != epoch { + return errWorkStopping + } + return nil +} + +// This is the last cancellable boundary before the existing atomic home merge. +// A stop that won earlier keeps the isolated copy. Once homecoming has begun, +// its actual receipt must survive; Stop only suppresses subsequent model work. +func (a *Agent) beginBeltHomecoming(run *beltRun) bool { + a.beltMu.Lock() + defer a.beltMu.Unlock() + if run.stopped { + return false + } + run.bringingHome = true + return true +} +func (a *Agent) beltRunQuiet(run *beltRun) bool { + a.beltMu.Lock() + defer a.beltMu.Unlock() + return run.quiet +} +func (a *Agent) settleBeltStopIfNeeded(run *beltRun, summary RunSummary) bool { + if stopped, why := a.beltRunStopped(run); stopped { + a.settleStoppedBeltRun(run, why, summary.Cut) + a.settleTaskCrew(run.row, router.CrewStopped, summary.USD) + return true + } + return false +} diff --git a/internal/session/task_run_belt.go b/internal/session/task_run_belt.go index 85ee77bb8..adcbdc273 100644 --- a/internal/session/task_run_belt.go +++ b/internal/session/task_run_belt.go @@ -39,6 +39,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "github.com/Agent-Field/codeaf/internal/config" @@ -390,6 +391,12 @@ type beltRun struct { cut context.CancelFunc stopped bool stopReason string + // bringingHome linearizes Stop against the existing main-checkout merge. + // quiet suppresses model narration after a stop arrives beyond that point. + // deliveryMu serializes page Stop with final receipt admission. + deliveryMu sync.Mutex + bringingHome bool + quiet bool // ending says the engine has answered and the run is only landing, // summarising and settling now: its supervisor is gone, so nothing will ever // run work added to its store. closing says the CONVERSATION is ending @@ -551,6 +558,11 @@ func (a *Agent) startOrJoinTaskRun(ctx context.Context, id uint64, title, brief // startOrJoinTaskRunVia is the one body behind every door above: the start // lock, the join, the folder or the copy, the store and the first row. func (a *Agent) startOrJoinTaskRunVia(ctx context.Context, id uint64, title, brief string, dependsOn []uint64, stand taskStand, question string, via *delegate.Delegate, asked ...string) (bool, error) { + epoch, err := a.beltAdmissionEpoch() + if err != nil { + return false, err + } + engine := chatRunEngine g := a.graph() if engine == nil || g == nil || g.planPath() == "" { @@ -597,6 +609,9 @@ func (a *Agent) startOrJoinTaskRunVia(ctx context.Context, id uint64, title, bri // as children, exactly as a hand-off made a minute later would. a.lockBeltStart() defer a.beltStartMu.Unlock() + if err := a.checkBeltAdmission(epoch); err != nil { + return false, err + } live, err := a.joinOrWait(ctx, stand, id, title, brief, dependencies, via) if err != nil { return false, err @@ -1607,6 +1622,13 @@ func (a *Agent) installBeltRun(g *TaskGraph, run *beltRun) { a.beltRun = run a.beltMu.Unlock() a.publishJoinedRunRows(g, run) + // Setup admitted before a stop inherits its closed admission gate. + a.mu.Lock() + stopped, why := a.workStopped, a.workStopReason + a.mu.Unlock() + if stopped { + _, _, _ = a.stopBeltRow(run.row, why) + } } // publishRunRow hands one run row to whoever is watching and keeps it for a @@ -1911,13 +1933,10 @@ func (a *Agent) driveBeltRun(ctx context.Context, engine RunEngine, run *beltRun if run.cut != nil { defer run.cut() } - if stopped, why := a.beltRunStopped(run); stopped { - // A RUN A PERSON STOPPED IS NOT LANDED. Its work is kept where the stop's - // own sentence said it would be, and the ending is the stop's (stoprun.go). - a.settleStoppedBeltRun(run, why, summary.Cut) - a.settleTaskCrew(run.row, router.CrewStopped, summary.USD) + if a.settleBeltStopIfNeeded(run, summary) { return } + if closing && run.delegate == nil && summary.Outcome != beltRunOutcomeDone { // THE CONVERSATION CLOSED UNDER THE RUN, AND THAT IS NOBODY'S ENDING. The // run is not landed, its row is not settled and nothing is written on its @@ -1932,42 +1951,19 @@ func (a *Agent) driveBeltRun(ctx context.Context, engine RunEngine, run *beltRun // branch below — never merged into the person's. return } - var landing RunLanding - if run.delegate != nil { - // A PROGRAM'S RUN IS OVER WHEN ITS PROGRAM IS, however it ended: it is - // a run of one task that nothing continues, so a store the engine left - // open — a program ended at a limit leaves it so — is closed here, or its - // page would read `running` and offer `stop it` for ever. A run that - // already ended is left as it ended. - // - // IT IS CLOSED BEFORE THE LANDING, NOT AFTER IT. The folder's last - // commit takes its time, and a page that went on reading `running` over - // a program that had already exited was a page claiming a present that - // was over — for the two limit endings alone, because every other ending - // is written by the engine at the program's exit. - if summary.Outcome != beltRunOutcomeDone { - words, _ := runEndingWords(summary) - _ = run.store.FailRoot(words) - } - landing = a.landDelegateRun(run, summary) - } else { - // EVERY OTHER ENDING IS WRITTEN ON THE RUN'S OWN TASK. The engine writes - // the ending of a limit or a failed root worker itself; this is the same - // write made again from the door, which the store takes once and ignores - // after, so no engine can leave a run the next hand-off would find still - // open. - if summary.Outcome != beltRunOutcomeDone { - _ = run.store.EndRoot(summary.Outcome) - } - landing = a.landBeltRun(ctx, engine, run) + landing := a.finishBeltRunLanding(ctx, engine, run, summary) + if a.settleBeltStopIfNeeded(run, summary) { + return } // A LANDING GETS ONE LAST READING before its digest is composed. The call // owns the short beltRunSummaryDeadline: refusal, malformed output, or a // slow provider leaves the stored reading alone and cannot hold the run // beyond that bound. RefreshRunSummary itself declines without a store. - refreshCtx, cancelRefresh := context.WithTimeout(withCrewTask(ctx, run.crew), beltRunSummaryDeadline) - a.RefreshRunSummary(refreshCtx, run.root, time.Time{}) - cancelRefresh() + if !a.beltRunQuiet(run) { + refreshCtx, cancelRefresh := context.WithTimeout(withCrewTask(ctx, run.crew), beltRunSummaryDeadline) + a.RefreshRunSummary(refreshCtx, run.root, time.Time{}) + cancelRefresh() + } note := beltRunOutcomeNote(run.store, run.root, summary, landing, a.beltRunSpan(run)) if run.delegate != nil { note = programPageNote(programName(run.delegate), landing) @@ -1993,6 +1989,42 @@ func (a *Agent) driveBeltRun(ctx context.Context, engine RunEngine, run *beltRun a.settleTaskCrew(run.row, outcome, run.crew.taskSpent(summary.USD)) } +// finishBeltRunLanding settles the isolated work before its receipt is published. +func (a *Agent) finishBeltRunLanding(ctx context.Context, engine RunEngine, run *beltRun, summary RunSummary) RunLanding { + var landing RunLanding + if run.delegate != nil { + // A PROGRAM'S RUN IS OVER WHEN ITS PROGRAM IS, however it ended: it is + // a run of one task that nothing continues, so a store the engine left + // open — a program ended at a limit leaves it so — is closed here, or its + // page would read `running` and offer `stop it` for ever. A run that + // already ended is left as it ended. + // + // IT IS CLOSED BEFORE THE LANDING, NOT AFTER IT. The folder's last + // commit takes its time, and a page that went on reading `running` over + // a program that had already exited was a page claiming a present that + // was over — for the two limit endings alone, because every other ending + // is written by the engine at the program's exit. + if summary.Outcome != beltRunOutcomeDone { + words, _ := runEndingWords(summary) + _ = run.store.FailRoot(words) + } + if a.beginBeltHomecoming(run) { + landing = a.landDelegateRun(run, summary) + } + } else { + // EVERY OTHER ENDING IS WRITTEN ON THE RUN'S OWN TASK. The engine writes + // the ending of a limit or a failed root worker itself; this is the same + // write made again from the door, which the store takes once and ignores + // after, so no engine can leave a run the next hand-off would find still + // open. + if summary.Outcome != beltRunOutcomeDone { + _ = run.store.EndRoot(summary.Outcome) + } + landing = a.landBeltRun(ctx, engine, run) + } + return landing +} + // crewKept is whether a run's ending counts as its crew's work kept: it // finished, and its landing was neither refused, conflicted nor abandoned. // @@ -2058,6 +2090,9 @@ func (a *Agent) releaseBeltRun(run *beltRun) { func (a *Agent) landBeltRun(ctx context.Context, engine RunEngine, run *beltRun) RunLanding { landing, err := engine.Land(ctx, run.store, run.workspace, run.tree.checkBase, run.root) if err != nil { + if stopped, _ := a.beltRunStopped(run); stopped { + return RunLanding{} + } if g := a.graph(); g != nil { g.planNote("the run's landing failed: " + err.Error()) } @@ -2071,6 +2106,9 @@ func (a *Agent) landBeltRun(ctx context.Context, engine RunEngine, run *beltRun) // across or the branch kept and the files named, the copy given back, and the // homecoming written on the run's page. func (a *Agent) bringBeltRunHome(run *beltRun, landing RunLanding) RunLanding { + if !a.beginBeltHomecoming(run) { + return landing + } if run.tree.dir == "" { return landing } @@ -2107,13 +2145,22 @@ func (a *Agent) bringBeltRunHome(run *beltRun, landing RunLanding) RunLanding { // deliverBeltRunLanding writes the run's digest into the conversation record. // A LANDING SPEAKS ONLY WHEN AN ANSWER IS OWED — or when a program ended it, // whose ending is always the conversation's to act on (program_outcome.go). +var beltLandingAdmissionWaits func() + func (a *Agent) deliverBeltRunLanding(run *beltRun, summary RunSummary, landing RunLanding) { line := beltRunOutcomeNote(run.store, run.root, summary, landing, a.beltRunSpan(run)) + if a.beltRunQuiet(run) { + a.recordQuietBeltLanding(run, summary, line, landing) + return + } + if beltLandingAdmissionWaits != nil { + beltLandingAdmissionWaits() + } if run.delegate != nil { if limit := programLimitLine(run, summary, landing); limit != "" { a.recordProgramLimit(limit) } - a.accept(delivery{origin: fromRuntime, kind: msgResult, note: a.programLandingNote(run, summary, line)}) + a.admitBeltLanding(run, summary, line, landing, a.programLandingNote(run, summary, line)) return } task := run.store.Task(run.root) @@ -2128,7 +2175,7 @@ func (a *Agent) deliverBeltRunLanding(run *beltRun, summary RunSummary, landing note.settle, note.settleCeiling = true, owedLandingCallCeiling() note.settlePrompt = landingAnswerPrompt note.settleModel, _ = roles.TierModel(roles.Source(a.config.RolesSource), owedLandingTier()) - a.accept(delivery{origin: fromRuntime, kind: msgResult, note: note}) + a.admitBeltLanding(run, summary, line, landing, note) return } note := userText(line) @@ -2140,6 +2187,38 @@ func (a *Agent) deliverBeltRunLanding(run *beltRun, summary RunSummary, landing a.mu.Unlock() } +// admitBeltLanding makes the last quiet check and wake admission one step +// against page Stop. Receipt construction and persistence happen outside it. +func (a *Agent) admitBeltLanding(run *beltRun, summary RunSummary, line string, landing RunLanding, note userMessage) { + run.deliveryMu.Lock() + if a.beltRunQuiet(run) { + run.deliveryMu.Unlock() + a.recordQuietBeltLanding(run, summary, line, landing) + return + } + a.accept(delivery{origin: fromRuntime, kind: msgResult, note: note}) + run.deliveryMu.Unlock() +} + +// recordQuietBeltLanding preserves the actual result and its operational +// attribution after a late stop, without admitting another model turn. +func (a *Agent) recordQuietBeltLanding(run *beltRun, summary RunSummary, line string, landing RunLanding) { + note := userText(line) + if run.delegate != nil { + if limit := programLimitLine(run, summary, landing); limit != "" { + a.recordProgramLimit(limit) + } + note = a.programLandingNote(run, summary, line) + } + note.wake = false + note.completionOnly = true + note.replyTags = plainBeltLandingReplyTags(run, run.store.Task(run.root), summary) + note.authored = true + a.mu.Lock() + a.recordUserLocked(note) + a.mu.Unlock() +} + // Plain run receipts retain their task identity while the original delivery and answer obligations stay unchanged. func plainBeltLandingReplyTags(run *beltRun, task *plandb.Task, summary RunSummary) []TaskReplyTag { id := run.row diff --git a/internal/session/task_run_continue.go b/internal/session/task_run_continue.go index a8f31fab1..2dea3b2f5 100644 --- a/internal/session/task_run_continue.go +++ b/internal/session/task_run_continue.go @@ -40,6 +40,10 @@ import ( // door that spends money says what it did in the words the person will read // about it afterwards. func (a *Agent) ContinueRun(ctx context.Context, row uint64) (string, error) { + epoch, err := a.beltAdmissionEpoch() + if err != nil { + return "", err + } engine := chatRunEngine g := a.graph() if engine == nil || g == nil || g.planPath() == "" { @@ -52,6 +56,9 @@ func (a *Agent) ContinueRun(ctx context.Context, row uint64) (string, error) { // store ([Agent.lockBeltStart]). a.lockBeltStart() defer a.beltStartMu.Unlock() + if err := a.checkBeltAdmission(epoch); err != nil { + return "", err + } a.beltMu.Lock() live := a.beltRun a.beltMu.Unlock() diff --git a/internal/session/task_run_stop_boundaries_test.go b/internal/session/task_run_stop_boundaries_test.go new file mode 100644 index 000000000..767af413d --- /dev/null +++ b/internal/session/task_run_stop_boundaries_test.go @@ -0,0 +1,196 @@ +package session + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/delegate" + "github.com/Agent-Field/codeaf/internal/plandb" +) + +func TestContinueWaitingBeforeStopCannotBorrowFreshSubmissionAdmission(t *testing.T) { + agent, _, _, double := continueAgent(t) + entered, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + releaseWaiter := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(releaseWaiter) + prior := beltStartWaits + beltStartWaits = func() { close(entered); <-release } + t.Cleanup(func() { beltStartWaits = prior }) + agent.beltStartMu.Lock() + result := make(chan error, 1) + go func() { _, err := agent.ContinueRun(context.Background(), 99); result <- err }() + select { + case <-entered: + case <-time.After(time.Second): + agent.beltStartMu.Unlock() + t.Fatal("continue did not wait at start admission") + } + if err := agent.StopWork(); err != nil { + agent.beltStartMu.Unlock() + t.Fatal(err) + } + agent.beltStartMu.Unlock() + agent.mu.Lock() + stopped := agent.workStopDone + agent.mu.Unlock() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("stop did not join pending start admission") + } + // This is the same admission reopening the explicit Submit path performs. + agent.mu.Lock() + err := agent.resumeWorkLocked() + agent.mu.Unlock() + if err != nil { + t.Fatal(err) + } + releaseWaiter() + select { + case err := <-result: + if !errors.Is(err, errWorkStopping) { + t.Fatalf("old continue request borrowed new admission: %v", err) + } + case <-time.After(time.Second): + t.Fatal("old continue request did not settle") + } + nothingStarted(t, agent, double) +} + +func TestStopBeforeHomecomingKeepsItsCancellationAuthority(t *testing.T) { + agent, _ := newTestAgent(t, &scriptedCompleter{}, nil) + ctx, cut := context.WithCancel(context.Background()) + run := &beltRun{row: 95, title: "fixture work", cut: cut} + // The authoritative stop flag is set by stopBeltRow before its context cut. + agent.beltMu.Lock() + run.stopped = true + agent.beltMu.Unlock() + if agent.beginBeltHomecoming(run) { + t.Fatal("a stopped isolated copy was admitted to the main checkout") + } + cut() + if ctx.Err() == nil { + t.Fatal("stop did not cut remaining calls") + } +} + +func TestStopAfterHomecomingBeganPreservesReceiptAndCutsFurtherCalls(t *testing.T) { + agent, _ := newTestAgent(t, &scriptedCompleter{}, nil) + ctx, cut := context.WithCancel(context.Background()) + run := &beltRun{row: 96, title: "fixture finished work", cut: cut} + agent.beltMu.Lock() + agent.beltRun = run + agent.beltMu.Unlock() + // Clear the synthetic registration before Agent.Close, which expects a store. + t.Cleanup(func() { agent.beltMu.Lock(); agent.beltRun = nil; agent.beltMu.Unlock() }) + if !agent.beginBeltHomecoming(run) { + t.Fatal("finished work could not begin its homecoming") + } + line, owned, err := agent.stopBeltRow(run.row, "") + if err != nil || !owned || !strings.Contains(line, "has already finished") { + t.Fatalf("stop promised to undo a homecoming already begun: %q %v", line, err) + } + if ctx.Err() == nil || !agent.beltRunQuiet(run) { + t.Fatal("post-homecoming stop left further narration able to spend") + } + agent.beltMu.Lock() + stopped := run.stopped + agent.beltMu.Unlock() + if stopped { + t.Fatal("post-homecoming stop rewrote the actual completion as an unlanded stopped branch") + } +} + +func TestQuietLandingKeepsOperationalTaskAndProgramProvenanceWithoutWake(t *testing.T) { + for _, program := range []*delegate.Delegate{nil, {Name: "fixture-review", Lands: delegate.LandsText}} { + name := "plain" + if program != nil { + name = program.Name + } + t.Run(name, func(t *testing.T) { + completer := &scriptedCompleter{} + agent, _ := newTestAgent(t, completer, nil) + store := askStore(t, "repair parser") + defer store.Close() + question := "What did the repair find?" + if _, err := store.Revise(store.RootID(), plandb.TaskPatch{Question: &question}); err != nil { + t.Fatal(err) + } + run := &beltRun{store: store, root: store.RootID(), row: 97, title: "Parser", brief: "repair parser", quiet: true, delegate: program} + agent.deliverBeltRunLanding(run, RunSummary{Outcome: beltRunOutcomeDone, Result: "Root output preserved."}, RunLanding{}) + entries := agent.Transcript() + if len(entries) != 1 || !entries[0].RoutineCompletion || len(entries[0].CompletionTasks) != 1 { + t.Fatalf("quiet receipt lost operational attribution: %+v", entries) + } + tag := entries[0].CompletionTasks[0] + if tag.ID != 97 || tag.Title != "Parser" || tag.Request != question || tag.State != TaskDone || !strings.Contains(entries[0].Text, "Root output preserved.") { + t.Fatalf("quiet receipt lost actual result or task: %+v", entries[0]) + } + agent.mu.Lock() + woke := agent.running || len(agent.steering) != 0 + hold := agent.programHold + agent.mu.Unlock() + if woke || completer.requests() != 0 { + t.Fatal("quiet receipt admitted another model turn") + } + if program != nil && (hold == nil || hold.row != run.row || hold.program != program.Name) { + t.Fatalf("quiet program receipt lost its provenance: %+v", hold) + } + }) + } +} + +func TestPageStopWinningFinalReceiptAdmissionDoesNotWake(t *testing.T) { + completer := &scriptedCompleter{steps: []step{finalText("Do not buy this answer.")}} + agent, _ := newTestAgent(t, completer, nil) + store := askStore(t, "repair parser") + defer store.Close() + question := "What did the repair find?" + if _, err := store.Revise(store.RootID(), plandb.TaskPatch{Question: &question}); err != nil { + t.Fatal(err) + } + run := &beltRun{store: store, root: store.RootID(), row: 98, title: "Parser", bringingHome: true} + agent.beltMu.Lock() + agent.beltRun = run + agent.beltMu.Unlock() + t.Cleanup(func() { agent.beltMu.Lock(); agent.beltRun = nil; agent.beltMu.Unlock() }) + entered, release, delivered := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + prior := beltLandingAdmissionWaits + beltLandingAdmissionWaits = func() { close(entered); <-release } + t.Cleanup(func() { unblock(); beltLandingAdmissionWaits = prior }) + go func() { + agent.deliverBeltRunLanding(run, RunSummary{Outcome: beltRunOutcomeDone, Result: "Preserved result."}, RunLanding{}) + close(delivered) + }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("receipt did not reach admission") + } + if _, owned, err := agent.stopBeltRow(run.row, ""); !owned || err != nil { + t.Fatalf("stop refused: owned=%v err=%v", owned, err) + } + unblock() + select { + case <-delivered: + case <-time.After(time.Second): + t.Fatal("receipt did not settle") + } + agent.mu.Lock() + woke := agent.running || len(agent.steering) != 0 + agent.mu.Unlock() + if woke || completer.requests() != 0 { + t.Fatal("receipt admitted a model wake after page Stop won") + } + entries := agent.Transcript() + if len(entries) != 1 || !entries[0].RoutineCompletion || len(entries[0].CompletionTasks) != 1 || entries[0].CompletionTasks[0].ID != run.row { + t.Fatalf("quiet receipt lost provenance: %+v", entries) + } +} diff --git a/internal/session/title.go b/internal/session/title.go index a6f2cf31b..6feb7fc6d 100644 --- a/internal/session/title.go +++ b/internal/session/title.go @@ -146,7 +146,7 @@ const titleWindow = 2 * time.Minute // shell opening has an ordinary answer. The gate and mark share the hold, so // two Submits racing cannot buy two names. func (a *Agent) startTitleLocked() { - if a.file == nil || a.titleTried || strings.TrimSpace(a.title) != "" || a.closed || a.titleCtx == nil { + if a.file == nil || a.titleTried || strings.TrimSpace(a.title) != "" || a.closed || a.workStopped || a.titleCtx == nil { return } // AND A TASK NODE NAMES NOTHING. A name exists for the picker, which lists diff --git a/internal/session/wallclock.go b/internal/session/wallclock.go index a17644aa5..81d260d86 100644 --- a/internal/session/wallclock.go +++ b/internal/session/wallclock.go @@ -42,10 +42,14 @@ func wallIsUp(steward *Steward) bool { // road leaves work alone, while this reader stops it and keeps what it did. const stopStoppedMovingTail = " · work was still going, so it was stopped and what it did was kept" -// armWallClock starts the one reader only for an unattended session whose -// budget names a wall. A person, an unattended session with no budget, and a -// money-only ceiling get no channel and no goroutine at all. +// armWallClock arms an explicit interactive launch wall, or the existing +// unattended Steward wall. Sessions without a wall and money-only ceilings +// get no channel or goroutine. func (a *Agent) armWallClock() { + if a.interactiveBudget().Wall > 0 { + a.armConversationWall() + return + } steward := a.steward() if steward == nil || steward.Budget().Wall <= 0 { return @@ -117,14 +121,7 @@ func (a *Agent) endRunOnTheWall() bool { } a.mu.Unlock() - graph := a.tasker() - if graph == nil { - return false - } - graph.mu.Lock() - flight := graph.flightLocked() - graph.mu.Unlock() - if len(flight.moving) == 0 { + if !a.ownedWorkIsMoving() { return false } @@ -148,9 +145,7 @@ func (a *Agent) endRunOnTheWall() bool { a.stopWorkOnTheWall() decision := stopSpent(why) - if len(flight.moving) > 0 { - decision.Reason += stopStoppedMovingTail - } + decision.Reason += stopStoppedMovingTail note := checkpointStoppedNote + decision.Reason a.journalDecision(decision) a.record(textMessage("assistant", note)) @@ -178,21 +173,56 @@ func (a *Agent) endRunOnTheWall() bool { // lets a running worker land its stopped row, and deliberately leaves every // branch and working copy alone. func (a *Agent) stopWorkOnTheWall() { - graph := a.tasker() - if graph == nil { + if err := a.stopWorkFor(conversationTimeLimitReason, StopByTimeLimit); err != nil { return } - graph.mu.Lock() - graph.quitting = true - ids := make([]uint64, 0, len(graph.order)) - for _, id := range graph.order { - if node := graph.nodes[id]; node != nil && !node.state.settled() { - ids = append(ids, id) - } + a.mu.Lock() + done := a.workStopDone + a.mu.Unlock() + if done != nil { + <-done } - graph.mu.Unlock() +} - for _, id := range ids { - _, _ = graph.stop(id) +// ownedWorkIsMoving observes the same producers StopWork owns. A standalone +// job or adaptive run need not have a moving row in the task graph. +func (a *Agent) ownedWorkIsMoving() bool { + graph := a.tasker() + if graph != nil { + graph.mu.Lock() + moving := len(graph.flightLocked().moving) > 0 + graph.mu.Unlock() + if moving { + return true + } + } + a.beltMu.Lock() + beltMoving := a.beltRun != nil + a.beltMu.Unlock() + if beltMoving { + return true + } + a.mu.Lock() + jobs, harnessMoving := a.jobs, len(a.harnessRuns) > 0 + runs := make([]*orchestration, 0, len(a.orchestrations)) + for _, run := range a.orchestrations { + runs = append(runs, run) + } + a.mu.Unlock() + if harnessMoving { + return true + } + if jobs != nil { + for _, job := range jobs.all() { + if job.running() { + return true + } + } + } + for _, live := range runs { + if !live.run.Snapshot().Done { + return true + } } + return false } diff --git a/internal/tui3/taskplan.go b/internal/tui3/taskplan.go index 6c8194a5e..af56b24e7 100644 --- a/internal/tui3/taskplan.go +++ b/internal/tui3/taskplan.go @@ -646,10 +646,11 @@ func planNamesOf(rows []tasksMineRow, chat string) map[string]bool { // landed yet, and the run's store is archived the moment a finished plan is // replaced ([session.Agent.openBeltRunStore]) — dropping a row on the strength // of an identity nothing answers for would take the run off the page -// altogether, which is worse than the row it replaces. -func planStoreDraws(rows []tasksMineRow, plan []session.PlanTaskRow) []tasksMineRow { +// altogether, which is worse than the row it replaces. The returned identities +// also suppress those same published rows from the persisted world scan. +func planStoreDraws(rows []tasksMineRow, plan []session.PlanTaskRow) ([]tasksMineRow, map[tasksKey]bool) { if len(rows) == 0 || len(plan) == 0 { - return rows + return rows, nil } held := make(map[string]bool, len(plan)) for _, task := range plan { @@ -658,13 +659,15 @@ func planStoreDraws(rows []tasksMineRow, plan []session.PlanTaskRow) []tasksMine } } kept := make([]tasksMineRow, 0, len(rows)) + replaced := make(map[tasksKey]bool) for _, row := range rows { if held[strings.TrimSpace(row.planTask)] { + replaced[tasksKeyOf(row.entry)] = true continue } kept = append(kept, row) } - return kept + return kept, replaced } // ── THE PAGE ONE PLAN ROW OPENS ───────────────────────────────────────────── diff --git a/internal/tui3/tasksplace.go b/internal/tui3/tasksplace.go index 9af954461..529d12b50 100644 --- a/internal/tui3/tasksplace.go +++ b/internal/tui3/tasksplace.go @@ -272,7 +272,8 @@ func readTasks(world session.World, mine tasksMine, win session.UsageWindow, by // done here rather than in either loop so the title dedupe further down sees // the same set of rows the page does — a run's row left in the name set // would hide its own plan row. - mine.rows = planStoreDraws(mine.rows, mine.plan) + var replaced map[tasksKey]bool + mine.rows, replaced = planStoreDraws(mine.rows, mine.plan) // order keeps the pass stable: a map alone would re-order the page on every // frame it was rebuilt, and the sections below are drawn in the order the // rows arrived within each one. @@ -288,6 +289,11 @@ func readTasks(world session.World, mine tasksMine, win session.UsageWindow, by for _, project := range world.Projects { for _, row := range project.Sessions { for _, entry := range row.Tasks.Rows { + // The live store replaces both the in-memory published row and + // that same identity from the project's persisted world scan. + if replaced[tasksKeyOf(entry)] { + continue + } // The world scan has already judged this row against the // conversation that wrote it (world.go's first law), so its // answer travels with it. diff --git a/internal/tui3/tasksplace_test.go b/internal/tui3/tasksplace_test.go index f7a265262..5fede192f 100644 --- a/internal/tui3/tasksplace_test.go +++ b/internal/tui3/tasksplace_test.go @@ -998,3 +998,39 @@ func TestAConversationWithNoTitleIsCalledTheWordNotItsId(t *testing.T) { } } } + +func TestLivePlanReplacesItsPublishedRecordAcrossTheWorldReading(t *testing.T) { + now := time.Now() + published := session.TaskIndexEntry{ID: "7", Title: "Mock counter", SessionID: "current-chat", Status: string(session.TaskRunning), StartedAt: now} + unrelated := published + unrelated.SessionID = "other-chat" + own := session.SessionRow{ID: "current-chat", Title: "Current", Tasks: session.TaskRollup{Rows: []session.TaskIndexEntry{published}}} + other := session.SessionRow{ID: "other-chat", Title: "Other", Tasks: session.TaskRollup{Rows: []session.TaskIndexEntry{unrelated}}} + world := session.World{Projects: []session.Project{{Sessions: []session.SessionRow{own, other}}}} + mine := tasksMine{row: own, rows: []tasksMineRow{{entry: published, planTask: "t-7"}}, plan: []session.PlanTaskRow{{ID: "t-7", Title: published.Title, Status: "claimed", Started: now}}} + reading := readTasks(world, mine, session.LastDays(now, 10), tasksSort{}, time.Time{}, now) + if len(reading.items) != 2 { + t.Fatalf("one live run and unrelated chat produced %d rows, want 2: %+v", len(reading.items), reading.items) + } + foundPlan, foundOther := false, false + for _, item := range reading.items { + switch item.entry.SessionID { + case own.ID: + foundPlan = item.plan != nil && item.plan.ID == "t-7" && item.status().Word == "running" + if !foundPlan { + t.Fatalf("live run opens a historical card instead of its stoppable plan room: %+v", item) + } + case other.ID: + foundOther = item.plan == nil && item.entry.ID == unrelated.ID + } + } + if !foundPlan || !foundOther { + t.Fatalf("store identity crossed conversation boundary: plan=%v other=%v", foundPlan, foundOther) + } + // An absent/still-loading store cannot erase the only recorded work. + mine.plan = nil + reading = readTasks(world, mine, session.LastDays(now, 10), tasksSort{}, time.Time{}, now) + if len(reading.items) != 2 { + t.Fatalf("missing plan erased its index record: %+v", reading.items) + } +}