From 3ca2ce6e4c3ca66884369ef54d7ebebcf719fc60 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:29:15 -0400 Subject: [PATCH 01/10] Fix stop admission and preserve completed landing receipts --- internal/session/stoprun.go | 9 ++ internal/session/task_run_admission.go | 54 +++++++ internal/session/task_run_belt.go | 130 +++++++++++----- internal/session/task_run_continue.go | 7 + .../session/task_run_stop_boundaries_test.go | 146 ++++++++++++++++++ 5 files changed, 310 insertions(+), 36 deletions(-) create mode 100644 internal/session/task_run_admission.go create mode 100644 internal/session/task_run_stop_boundaries_test.go diff --git a/internal/session/stoprun.go b/internal/session/stoprun.go index b78455d685..cd0d157618 100644 --- a/internal/session/stoprun.go +++ b/internal/session/stoprun.go @@ -79,6 +79,15 @@ func (a *Agent) stopBeltRow(id uint64, why string) (string, bool, error) { } if run.row == id { name := taskStopName(id, run.title) + if run.bringingHome { + run.quiet = true + cut := run.cut + a.beltMu.Unlock() + if cut != nil { + cut() + } + return name + " has already finished; there is nothing to stop", true, nil + } if run.stopped { a.beltMu.Unlock() return name + " is already stopping", true, nil diff --git a/internal/session/task_run_admission.go b/internal/session/task_run_admission.go new file mode 100644 index 0000000000..c30b8c73f1 --- /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 85ee77bb8c..b077b8a069 100644 --- a/internal/session/task_run_belt.go +++ b/internal/session/task_run_belt.go @@ -390,6 +390,10 @@ 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. + 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 +555,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 +606,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 +1619,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 +1930,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 +1948,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 +1986,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 +2087,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 +2103,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 } @@ -2109,6 +2144,10 @@ func (a *Agent) bringBeltRunHome(run *beltRun, landing RunLanding) RunLanding { // whose ending is always the conversation's to act on (program_outcome.go). 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 run.delegate != nil { if limit := programLimitLine(run, summary, landing); limit != "" { a.recordProgramLimit(limit) @@ -2140,6 +2179,25 @@ func (a *Agent) deliverBeltRunLanding(run *beltRun, summary RunSummary, landing a.mu.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 a8f31fab10..2dea3b2f5b 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 0000000000..62a863594e --- /dev/null +++ b/internal/session/task_run_stop_boundaries_test.go @@ -0,0 +1,146 @@ +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) + } + }) + } +} From 0870c17bad48345d202ceded0b0722dbb174a7ad Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:33:32 -0400 Subject: [PATCH 02/10] fix(session): stop owned descendants and helpers at conversation limits --- internal/manual/chat/task-controls.md | 18 +++- internal/session/conversation_wall.go | 64 ++++++++++++ internal/session/handlepick.go | 8 +- internal/session/interactive_budget.go | 8 +- internal/session/memory.go | 2 +- internal/session/session.go | 9 +- internal/session/stop_auxiliary_test.go | 79 +++++++++++++++ internal/session/stop_descendants_test.go | 113 ++++++++++++++++++++++ internal/session/stopcause.go | 4 + internal/session/stopwork.go | 68 +++++++++++-- internal/session/title.go | 2 +- internal/session/wallclock.go | 82 +++++++++++----- 12 files changed, 414 insertions(+), 43 deletions(-) create mode 100644 internal/session/conversation_wall.go create mode 100644 internal/session/stop_auxiliary_test.go create mode 100644 internal/session/stop_descendants_test.go diff --git a/internal/manual/chat/task-controls.md b/internal/manual/chat/task-controls.md index 1bff6947b5..e393d9f88e 100644 --- a/internal/manual/chat/task-controls.md +++ b/internal/manual/chat/task-controls.md @@ -63,13 +63,29 @@ is kept; a design task can discard its unsaved page, and the confirmation says which consequence applies. Runs and background jobs use their own existing stop confirmations. The header and controls stop the work on the open page. +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. + +## 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. + ## Asking the chat to stop a task — can I tell the chat to stop a task, I asked it to stop task 2 and it kept going, the model said it stopped but the task is still running, cancel task 2 **Yes, and it is a real stop.** Say "stop task 2", "cancel task 2" or "drop task 2" in the 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/session/conversation_wall.go b/internal/session/conversation_wall.go new file mode 100644 index 0000000000..9c004d158b --- /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 5d5ff3daa3..761880daa9 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 c4a805a302..909831c77b 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 feb70480b4..81e62e24a3 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 439bacef18..d86917a60f 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 0000000000..6c9b2ac583 --- /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 0000000000..286ae1cf51 --- /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 a9e1f365b4..93f4129cc6 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/stopwork.go b/internal/session/stopwork.go index 8b3958ab1d..6252718639 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/title.go b/internal/session/title.go index a6f2cf31b6..6feb7fc6d1 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 a17644aa5f..81d260d868 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 } From 6416ee1a3e3f56f85cb260b0e03b933e166b1b78 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:37:01 -0400 Subject: [PATCH 03/10] Serialize final run receipt admission with task Stop --- internal/session/stoprun.go | 34 ++++++++++--- internal/session/task_run_belt.go | 25 +++++++++- .../session/task_run_stop_boundaries_test.go | 50 +++++++++++++++++++ 3 files changed, 99 insertions(+), 10 deletions(-) diff --git a/internal/session/stoprun.go b/internal/session/stoprun.go index cd0d157618..f98279d757 100644 --- a/internal/session/stoprun.go +++ b/internal/session/stoprun.go @@ -71,30 +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 - a.beltMu.Unlock() + 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 @@ -117,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/task_run_belt.go b/internal/session/task_run_belt.go index b077b8a069..adcbdc2739 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" @@ -392,6 +393,8 @@ type beltRun struct { 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, @@ -2142,17 +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) @@ -2167,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) @@ -2179,6 +2187,19 @@ 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) { diff --git a/internal/session/task_run_stop_boundaries_test.go b/internal/session/task_run_stop_boundaries_test.go index 62a863594e..767af413df 100644 --- a/internal/session/task_run_stop_boundaries_test.go +++ b/internal/session/task_run_stop_boundaries_test.go @@ -144,3 +144,53 @@ func TestQuietLandingKeepsOperationalTaskAndProgramProvenanceWithoutWake(t *test }) } } + +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) + } +} From 34e9dcd154896ab39d27e338d6c2299288a058dd Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:39:21 -0400 Subject: [PATCH 04/10] test: verify terminal stop and time limits end owned task work --- .../e2e/stop_descendants_wire_e2e_test.go | 280 ++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 internal/e2e/stop_descendants_wire_e2e_test.go 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 0000000000..2da20040c4 --- /dev/null +++ b/internal/e2e/stop_descendants_wire_e2e_test.go @@ -0,0 +1,280 @@ +//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" { + rig.lit("/history") + rig.keys("Enter") + time.Sleep(500 * time.Millisecond) + rig.keys("Right") + rig.keys("Down") + rig.keys("Enter") + time.Sleep(500 * time.Millisecond) + rig.keys("x") + rig.waitFor(10*time.Second, "Stop this task?") + rig.keys("1") + rig.keys("Enter") + } + if mode == "conversation_wall" { + rig.waitFor(35*time.Second, "conversation time limit reached") + rig.waitFor(10*time.Second, "a time limit you set stopped it") + } 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) + } + rig.lit("/history") + rig.keys("Enter") + time.Sleep(500 * time.Millisecond) + rig.keys("Right") + rig.keys("Down") + rig.keys("Enter") + time.Sleep(500 * time.Millisecond) + rig.keys("x") + rig.waitFor(10*time.Second, "Stop this task?") + rig.keys("1") + rig.keys("Enter") + 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 +} From 0b141f16fdc2de5179ddd96881cc87944feb0db7 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:44:52 -0400 Subject: [PATCH 05/10] docs: place stop and wall guidance in matching help sections --- internal/manual/chat/running-from-the-terminal.md | 9 +++++++++ internal/manual/chat/task-controls.md | 15 --------------- internal/manual/chat/worker-harness.md | 6 ++++++ 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/internal/manual/chat/running-from-the-terminal.md b/internal/manual/chat/running-from-the-terminal.md index 2c2b398e24..d74574744c 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 e393d9f88e..71fc8345a6 100644 --- a/internal/manual/chat/task-controls.md +++ b/internal/manual/chat/task-controls.md @@ -63,21 +63,6 @@ is kept; a design task can discard its unsaved page, and the confirmation says which consequence applies. Runs and background jobs use their own existing stop confirmations. The header and controls stop the work on the open page. -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. - -## 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. - ## Asking the chat to stop a task — can I tell the chat to stop a task, I asked it to stop task 2 and it kept going, the model said it stopped but the task is still running, cancel task 2 **Yes, and it is a real stop.** Say "stop task 2", "cancel task 2" or "drop task 2" in the diff --git a/internal/manual/chat/worker-harness.md b/internal/manual/chat/worker-harness.md index d20e5c891d..250a282ae1 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 From 171bc4f9a2dccc19113df0959462499dfa9116bf Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:47:54 -0400 Subject: [PATCH 06/10] test: wait for the active task room and disclose stop reasons --- .../e2e/stop_descendants_wire_e2e_test.go | 54 +++++++++++-------- 1 file changed, 31 insertions(+), 23 deletions(-) diff --git a/internal/e2e/stop_descendants_wire_e2e_test.go b/internal/e2e/stop_descendants_wire_e2e_test.go index 2da20040c4..dff021664f 100644 --- a/internal/e2e/stop_descendants_wire_e2e_test.go +++ b/internal/e2e/stop_descendants_wire_e2e_test.go @@ -105,21 +105,17 @@ func TestTerminalStopEndsOwnedTasksOnTheRealTerminal(t *testing.T) { } } if mode != "conversation_wall" { - rig.lit("/history") - rig.keys("Enter") - time.Sleep(500 * time.Millisecond) - rig.keys("Right") - rig.keys("Down") - rig.keys("Enter") - time.Sleep(500 * time.Millisecond) - rig.keys("x") - rig.waitFor(10*time.Second, "Stop this task?") - rig.keys("1") - rig.keys("Enter") + stopOwnedTaskThroughItsPage(t, rig) } if mode == "conversation_wall" { rig.waitFor(35*time.Second, "conversation time limit reached") - rig.waitFor(10*time.Second, "a time limit you set stopped it") + if screen := rig.capture(); strings.Contains(screen, "background work") && strings.Contains(screen, "ctrl+e") { + rig.keys("C-e") + } + screen := rig.waitFor(10*time.Second, "a time limit you set stopped it") + if !strings.Contains(screen, "a time limit you set stopped it") { + t.Fatal("joined wall stop reason was not disclosed") + } } else { rig.waitFor(20*time.Second, "stopped") } @@ -197,17 +193,7 @@ func TestTerminalStopLiveOwnedTaskWorkflow(t *testing.T) { } time.Sleep(pollEvery) } - rig.lit("/history") - rig.keys("Enter") - time.Sleep(500 * time.Millisecond) - rig.keys("Right") - rig.keys("Down") - rig.keys("Enter") - time.Sleep(500 * time.Millisecond) - rig.keys("x") - rig.waitFor(10*time.Second, "Stop this task?") - rig.keys("1") - rig.keys("Enter") + stopOwnedTaskThroughItsPage(t, rig) rig.waitFor(20*time.Second, "stopped") bytesAtStop := stopHeartbeatBytes(home) + stopHeartbeatBytes(ws) time.Sleep(10 * time.Second) @@ -278,3 +264,25 @@ func stopSpendSnapshot(t *testing.T, home string) stopSpend { } return total } + +func stopOwnedTaskThroughItsPage(t *testing.T, rig *rig) { + t.Helper() + openTasksPlace(t, rig) + tasksPlaceRunRow(t, rig) + hit, screen := rig.waitForAny(15*time.Second, say(t, "tasksEnterInsideWord"), say(t, "tasksEnterRoomWord")) + if hit == "" { + t.Fatalf("active task row never offered its room: %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") +} From 9e4d4ab6bfd172e6f88bd0374b3eb2a0dee1ea0b Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:48:33 -0400 Subject: [PATCH 07/10] docs: record owned work cancellation fix --- docs/changes/unreleased/1907-stop-owned-work.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 docs/changes/unreleased/1907-stop-owned-work.md 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 0000000000..9ac6243d50 --- /dev/null +++ b/docs/changes/unreleased/1907-stop-owned-work.md @@ -0,0 +1,13 @@ +--- +kind: fixed +title: Stop delegated work and helpers at conversation limits +pr: 1907 +surface: [chat, engine] +invalidates: [] +--- + +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. From fd925494af9e63a8fc777b5e641a788591f85777 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:54:43 -0400 Subject: [PATCH 08/10] test: verify one live task and inspect its actual stop controls --- .../e2e/stop_descendants_wire_e2e_test.go | 59 ++++++++++++++++--- 1 file changed, 52 insertions(+), 7 deletions(-) diff --git a/internal/e2e/stop_descendants_wire_e2e_test.go b/internal/e2e/stop_descendants_wire_e2e_test.go index dff021664f..a52bebf1dc 100644 --- a/internal/e2e/stop_descendants_wire_e2e_test.go +++ b/internal/e2e/stop_descendants_wire_e2e_test.go @@ -112,9 +112,12 @@ func TestTerminalStopEndsOwnedTasksOnTheRealTerminal(t *testing.T) { if screen := rig.capture(); strings.Contains(screen, "background work") && strings.Contains(screen, "ctrl+e") { rig.keys("C-e") } - screen := rig.waitFor(10*time.Second, "a time limit you set stopped it") - if !strings.Contains(screen, "a time limit you set stopped it") { - t.Fatal("joined wall stop reason was not disclosed") + 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") @@ -268,11 +271,25 @@ func stopSpendSnapshot(t *testing.T, home string) stopSpend { func stopOwnedTaskThroughItsPage(t *testing.T, rig *rig) { t.Helper() openTasksPlace(t, rig) - tasksPlaceRunRow(t, rig) - hit, screen := rig.waitForAny(15*time.Second, say(t, "tasksEnterInsideWord"), say(t, "tasksEnterRoomWord")) - if hit == "" { - t.Fatalf("active task row never offered its room: %s", screen) + 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")) { @@ -286,3 +303,31 @@ func stopOwnedTaskThroughItsPage(t *testing.T, rig *rig) { 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 +} From 436de6fcb182a9ef8b07479bfbb82f31b87168ce Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:55:40 -0400 Subject: [PATCH 09/10] Show the live plan once when task history also has its record --- internal/tui3/taskplan.go | 11 ++++++---- internal/tui3/tasksplace.go | 8 ++++++- internal/tui3/tasksplace_test.go | 36 ++++++++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/internal/tui3/taskplan.go b/internal/tui3/taskplan.go index 6c8194a5eb..af56b24e71 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 9af9544613..529d12b50d 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 f7a265262a..5fede192fa 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) + } +} From 26f8df08207d4fe3d58b379b81996b4412f01a51 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Thu, 22 Oct 2026 22:59:02 -0400 Subject: [PATCH 10/10] docs: clarify stop guarantees and live task history fix --- docs/changes/unreleased/1907-stop-owned-work.md | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/docs/changes/unreleased/1907-stop-owned-work.md b/docs/changes/unreleased/1907-stop-owned-work.md index 9ac6243d50..4352efd0eb 100644 --- a/docs/changes/unreleased/1907-stop-owned-work.md +++ b/docs/changes/unreleased/1907-stop-owned-work.md @@ -3,7 +3,9 @@ kind: fixed title: Stop delegated work and helpers at conversation limits pr: 1907 surface: [chat, engine] -invalidates: [] +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 @@ -11,3 +13,6 @@ 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.