fix(agent): harden goal result persistence
- commit
- 8cf6b9ac7db1da007d1e03e5a7e352e219e77893
- author
- brice · · 2026-08-09 21:16 UTC
- parent
- 7e9841f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
11 files · +264 -77
M packages/agent/cli.go +22 -19
| @@ -688,28 +688,22 @@ func copyExtensionStates(states map[string]json.RawMessage) map[string]json.RawM | |||
| 688 | 688 | return out | |
| 689 | 689 | } | |
| 690 | 690 | ||
| 691 | - | func persistExtensionToolResult(mgr *extensions.Manager, sess *core.Session, result core.ToolResult) { | |
| 691 | + | func persistToolResultState(mgr *extensions.Manager, sess *core.Session, result core.ToolResult) error { | |
| 692 | 692 | if sess == nil { | |
| 693 | - | return | |
| 693 | + | return nil | |
| 694 | 694 | } | |
| 695 | 695 | if details, ok := result.Details.(extensions.ToolResultDetails); ok && details.Extension != "" && len(details.State) > 0 { | |
| 696 | 696 | if err := sess.AppendExtensionState(details.Extension, details.State); err != nil { | |
| 697 | - | fmt.Fprintf(os.Stderr, "extension state: %v\n", err) | |
| 698 | - | } else if mgr != nil { | |
| 697 | + | return fmt.Errorf("extension state: %w", err) | |
| 698 | + | } | |
| 699 | + | if mgr != nil { | |
| 699 | 700 | mgr.UpdateSessionState(details.Extension, details.State) | |
| 700 | 701 | } | |
| 701 | 702 | } | |
| 702 | 703 | if err := persistGoalToolResult(sess, result); err != nil { | |
| 703 | - | fmt.Fprintf(os.Stderr, "goal state: %v\n", err) | |
| 704 | + | return fmt.Errorf("goal state: %w", err) | |
| 704 | 705 | } | |
| 705 | - | } | |
| 706 | - | ||
| 707 | - | func persistToolResultForCurrentSession(sessionTransitionMu *sync.RWMutex, persistMu *sync.Mutex, currentSession func() *core.Session, mgr *extensions.Manager, result core.ToolResult) { | |
| 708 | - | sessionTransitionMu.RLock() | |
| 709 | - | defer sessionTransitionMu.RUnlock() | |
| 710 | - | persistMu.Lock() | |
| 711 | - | defer persistMu.Unlock() | |
| 712 | - | persistExtensionToolResult(mgr, currentSession(), result) | |
| 706 | + | return nil | |
| 713 | 707 | } | |
| 714 | 708 | ||
| 715 | 709 | func persistGoalToolResult(sess *core.Session, result core.ToolResult) error { | |
| @@ -990,7 +984,7 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 990 | 984 | return err | |
| 991 | 985 | } | |
| 992 | 986 | r.Provider, r.Model = providerName, model | |
| 993 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | |
| 987 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } | |
| 994 | 988 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 995 | 989 | } | |
| 996 | 990 | announceSession(extMgr, sess) | |
| @@ -1071,7 +1065,7 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||
| 1071 | 1065 | return err | |
| 1072 | 1066 | } | |
| 1073 | 1067 | r.Provider, r.Model = providerName, model | |
| 1074 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | |
| 1068 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } | |
| 1075 | 1069 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 1076 | 1070 | } | |
| 1077 | 1071 | announceSession(extMgr, sess) | |
| @@ -1191,7 +1185,7 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||
| 1191 | 1185 | return err | |
| 1192 | 1186 | } | |
| 1193 | 1187 | r.Provider, r.Model = providerName, model | |
| 1194 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | |
| 1188 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } | |
| 1195 | 1189 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | |
| 1196 | 1190 | } | |
| 1197 | 1191 | announceSession(extMgr, sess) | |
| @@ -1842,8 +1836,15 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1842 | 1836 | goal := *sess.Meta.Goal | |
| 1843 | 1837 | return &goal | |
| 1844 | 1838 | } | |
| 1845 | - | persistToolResult := func(_ string, result core.ToolResult) { | |
| 1846 | - | persistToolResultForCurrentSession(&sessionTransitionMu, &persistMu, func() *core.Session { return sess }, extMgr, result) | |
| 1839 | + | bindAgentToolResultSession := func(a *core.Agent, boundSession *core.Session) { | |
| 1840 | + | if a == nil { | |
| 1841 | + | return | |
| 1842 | + | } | |
| 1843 | + | a.CommitToolResult = func(_ string, result core.ToolResult) error { | |
| 1844 | + | persistMu.Lock() | |
| 1845 | + | defer persistMu.Unlock() | |
| 1846 | + | return persistToolResultState(extMgr, boundSession, result) | |
| 1847 | + | } | |
| 1847 | 1848 | } | |
| 1848 | 1849 | wireAgentPersist := func(a *core.Agent) *core.Agent { | |
| 1849 | 1850 | if a == nil { | |
| @@ -1858,7 +1859,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1858 | 1859 | a.OnMessageAppended = persistMessage | |
| 1859 | 1860 | a.OnUsage = persistUsage | |
| 1860 | 1861 | a.OnTranscriptCompacted = persistCompaction | |
| 1861 | - | a.OnToolResult = persistToolResult | |
| 1862 | + | bindAgentToolResultSession(a, currentSession) | |
| 1862 | 1863 | return a | |
| 1863 | 1864 | } | |
| 1864 | 1865 | wireAgentPersist(ag) | |
| @@ -1967,6 +1968,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1967 | 1968 | oldCloseErr = oldSess.Close() | |
| 1968 | 1969 | } | |
| 1969 | 1970 | sess = candidate.session | |
| 1971 | + | bindAgentToolResultSession(candidate.agent, candidate.session) | |
| 1970 | 1972 | candidate.agent.SetSessionTimeContext(candidate.session.Meta.Started, candidate.session.Meta.Timezone, candidate.session.Meta.TimezoneOffset) | |
| 1971 | 1973 | sessionTitlePending = candidate.session != nil && candidate.session.Meta.Parent != "" && candidate.session.Title == "" && candidate.fullMessageCount <= candidate.session.Meta.ForkPoint | |
| 1972 | 1974 | // The live agent only receives a compact resume window, but the | |
| @@ -2169,6 +2171,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 2169 | 2171 | oldCloseErr = sess.Close() | |
| 2170 | 2172 | } | |
| 2171 | 2173 | sess = newSess | |
| 2174 | + | bindAgentToolResultSession(newAg, newSess) | |
| 2172 | 2175 | if newSess != nil { | |
| 2173 | 2176 | newAg.SetSessionTimeContext(newSess.Meta.Started, newSess.Meta.Timezone, newSess.Meta.TimezoneOffset) | |
| 2174 | 2177 | newStates = copyExtensionStates(newSess.ExtensionState) | |
| @@ -688,28 +688,22 @@ func copyExtensionStates(states map[string]json.RawMessage) map[string]json.RawM | |||||
| 688 | return out | 688 | return out | ||
| 689 | } | 689 | } | ||
| 690 | 690 | ||||
| 691 | - | func persistExtensionToolResult(mgr *extensions.Manager, sess *core.Session, result core.ToolResult) { | 691 | + | func persistToolResultState(mgr *extensions.Manager, sess *core.Session, result core.ToolResult) error { |
| 692 | if sess == nil { | 692 | if sess == nil { | ||
| 693 | - | return | 693 | + | return nil |
| 694 | } | 694 | } | ||
| 695 | if details, ok := result.Details.(extensions.ToolResultDetails); ok && details.Extension != "" && len(details.State) > 0 { | 695 | if details, ok := result.Details.(extensions.ToolResultDetails); ok && details.Extension != "" && len(details.State) > 0 { | ||
| 696 | if err := sess.AppendExtensionState(details.Extension, details.State); err != nil { | 696 | if err := sess.AppendExtensionState(details.Extension, details.State); err != nil { | ||
| 697 | - | fmt.Fprintf(os.Stderr, "extension state: %v\n", err) | 697 | + | return fmt.Errorf("extension state: %w", err) |
| 698 | - | } else if mgr != nil { | 698 | + | } |
| 699 | + | if mgr != nil { | |||
| 699 | mgr.UpdateSessionState(details.Extension, details.State) | 700 | mgr.UpdateSessionState(details.Extension, details.State) | ||
| 700 | } | 701 | } | ||
| 701 | } | 702 | } | ||
| 702 | if err := persistGoalToolResult(sess, result); err != nil { | 703 | if err := persistGoalToolResult(sess, result); err != nil { | ||
| 703 | - | fmt.Fprintf(os.Stderr, "goal state: %v\n", err) | 704 | + | return fmt.Errorf("goal state: %w", err) |
| 704 | } | 705 | } | ||
| 705 | - | } | 706 | + | return nil |
| 706 | - | ||||
| 707 | - | func persistToolResultForCurrentSession(sessionTransitionMu *sync.RWMutex, persistMu *sync.Mutex, currentSession func() *core.Session, mgr *extensions.Manager, result core.ToolResult) { | |||
| 708 | - | sessionTransitionMu.RLock() | |||
| 709 | - | defer sessionTransitionMu.RUnlock() | |||
| 710 | - | persistMu.Lock() | |||
| 711 | - | defer persistMu.Unlock() | |||
| 712 | - | persistExtensionToolResult(mgr, currentSession(), result) | |||
| 713 | } | 707 | } | ||
| 714 | 708 | ||||
| 715 | func persistGoalToolResult(sess *core.Session, result core.ToolResult) error { | 709 | func persistGoalToolResult(sess *core.Session, result core.ToolResult) error { | ||
| @@ -990,7 +984,7 @@ func runPrintMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 990 | return err | 984 | return err | ||
| 991 | } | 985 | } | ||
| 992 | r.Provider, r.Model = providerName, model | 986 | r.Provider, r.Model = providerName, model | ||
| 993 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | 987 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } |
| 994 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 988 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 995 | } | 989 | } | ||
| 996 | announceSession(extMgr, sess) | 990 | announceSession(extMgr, sess) | ||
| @@ -1071,7 +1065,7 @@ func runStreamMode(ctx context.Context, args Args, version string) (runErr error | |||||
| 1071 | return err | 1065 | return err | ||
| 1072 | } | 1066 | } | ||
| 1073 | r.Provider, r.Model = providerName, model | 1067 | r.Provider, r.Model = providerName, model | ||
| 1074 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | 1068 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } |
| 1075 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 1069 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 1076 | } | 1070 | } | ||
| 1077 | announceSession(extMgr, sess) | 1071 | announceSession(extMgr, sess) | ||
| @@ -1191,7 +1185,7 @@ func runJSONMode(ctx context.Context, args Args, version string) (runErr error) | |||||
| 1191 | return err | 1185 | return err | ||
| 1192 | } | 1186 | } | ||
| 1193 | r.Provider, r.Model = providerName, model | 1187 | r.Provider, r.Model = providerName, model | ||
| 1194 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | 1188 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } |
| 1195 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | 1189 | defer func() { runErr = joinSessionCloseError(runErr, sess) }() | ||
| 1196 | } | 1190 | } | ||
| 1197 | announceSession(extMgr, sess) | 1191 | announceSession(extMgr, sess) | ||
| @@ -1842,8 +1836,15 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1842 | goal := *sess.Meta.Goal | 1836 | goal := *sess.Meta.Goal | ||
| 1843 | return &goal | 1837 | return &goal | ||
| 1844 | } | 1838 | } | ||
| 1845 | - | persistToolResult := func(_ string, result core.ToolResult) { | 1839 | + | bindAgentToolResultSession := func(a *core.Agent, boundSession *core.Session) { |
| 1846 | - | persistToolResultForCurrentSession(&sessionTransitionMu, &persistMu, func() *core.Session { return sess }, extMgr, result) | 1840 | + | if a == nil { |
| 1841 | + | return | |||
| 1842 | + | } | |||
| 1843 | + | a.CommitToolResult = func(_ string, result core.ToolResult) error { | |||
| 1844 | + | persistMu.Lock() | |||
| 1845 | + | defer persistMu.Unlock() | |||
| 1846 | + | return persistToolResultState(extMgr, boundSession, result) | |||
| 1847 | + | } | |||
| 1847 | } | 1848 | } | ||
| 1848 | wireAgentPersist := func(a *core.Agent) *core.Agent { | 1849 | wireAgentPersist := func(a *core.Agent) *core.Agent { | ||
| 1849 | if a == nil { | 1850 | if a == nil { | ||
| @@ -1858,7 +1859,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1858 | a.OnMessageAppended = persistMessage | 1859 | a.OnMessageAppended = persistMessage | ||
| 1859 | a.OnUsage = persistUsage | 1860 | a.OnUsage = persistUsage | ||
| 1860 | a.OnTranscriptCompacted = persistCompaction | 1861 | a.OnTranscriptCompacted = persistCompaction | ||
| 1861 | - | a.OnToolResult = persistToolResult | 1862 | + | bindAgentToolResultSession(a, currentSession) |
| 1862 | return a | 1863 | return a | ||
| 1863 | } | 1864 | } | ||
| 1864 | wireAgentPersist(ag) | 1865 | wireAgentPersist(ag) | ||
| @@ -1967,6 +1968,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1967 | oldCloseErr = oldSess.Close() | 1968 | oldCloseErr = oldSess.Close() | ||
| 1968 | } | 1969 | } | ||
| 1969 | sess = candidate.session | 1970 | sess = candidate.session | ||
| 1971 | + | bindAgentToolResultSession(candidate.agent, candidate.session) | |||
| 1970 | candidate.agent.SetSessionTimeContext(candidate.session.Meta.Started, candidate.session.Meta.Timezone, candidate.session.Meta.TimezoneOffset) | 1972 | candidate.agent.SetSessionTimeContext(candidate.session.Meta.Started, candidate.session.Meta.Timezone, candidate.session.Meta.TimezoneOffset) | ||
| 1971 | sessionTitlePending = candidate.session != nil && candidate.session.Meta.Parent != "" && candidate.session.Title == "" && candidate.fullMessageCount <= candidate.session.Meta.ForkPoint | 1973 | sessionTitlePending = candidate.session != nil && candidate.session.Meta.Parent != "" && candidate.session.Title == "" && candidate.fullMessageCount <= candidate.session.Meta.ForkPoint | ||
| 1972 | // The live agent only receives a compact resume window, but the | 1974 | // The live agent only receives a compact resume window, but the | ||
| @@ -2169,6 +2171,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 2169 | oldCloseErr = sess.Close() | 2171 | oldCloseErr = sess.Close() | ||
| 2170 | } | 2172 | } | ||
| 2171 | sess = newSess | 2173 | sess = newSess | ||
| 2174 | + | bindAgentToolResultSession(newAg, newSess) | |||
| 2172 | if newSess != nil { | 2175 | if newSess != nil { | ||
| 2173 | newAg.SetSessionTimeContext(newSess.Meta.Started, newSess.Meta.Timezone, newSess.Meta.TimezoneOffset) | 2176 | newAg.SetSessionTimeContext(newSess.Meta.Started, newSess.Meta.Timezone, newSess.Meta.TimezoneOffset) | ||
| 2174 | newStates = copyExtensionStates(newSess.ExtensionState) | 2177 | newStates = copyExtensionStates(newSess.ExtensionState) | ||
M packages/agent/cli_session_test.go +24 -41
| @@ -7,7 +7,6 @@ import ( | |||
| 7 | 7 | "fmt" | |
| 8 | 8 | "os" | |
| 9 | 9 | "path/filepath" | |
| 10 | - | "runtime" | |
| 11 | 10 | "strings" | |
| 12 | 11 | "sync" | |
| 13 | 12 | "testing" | |
| @@ -20,7 +19,7 @@ import ( | |||
| 20 | 19 | "github.com/google/uuid" | |
| 21 | 20 | ) | |
| 22 | 21 | ||
| 23 | - | func TestPersistToolResultFinishesBeforeSessionTransition(t *testing.T) { | |
| 22 | + | func TestToolResultPersistenceUsesBoundSession(t *testing.T) { | |
| 24 | 23 | root := t.TempDir() | |
| 25 | 24 | original, err := core.NewSession(root, "original", "provider", "model", "test") | |
| 26 | 25 | if err != nil { | |
| @@ -38,53 +37,37 @@ func TestPersistToolResultFinishesBeforeSessionTransition(t *testing.T) { | |||
| 38 | 37 | } | |
| 39 | 38 | } | |
| 40 | 39 | ||
| 41 | - | var transitionMu sync.RWMutex | |
| 42 | - | var persistMu sync.Mutex | |
| 43 | - | current := original | |
| 40 | + | boundSession := original | |
| 41 | + | currentSession := next // Simulate a completed session switch before callback delivery. | |
| 44 | 42 | result := core.ToolResult{Details: tools.GoalUpdate{Status: core.GoalDone}} | |
| 45 | - | persistMu.Lock() | |
| 46 | - | persisted := make(chan struct{}) | |
| 47 | - | go func() { | |
| 48 | - | persistToolResultForCurrentSession(&transitionMu, &persistMu, func() *core.Session { return current }, nil, result) | |
| 49 | - | close(persisted) | |
| 50 | - | }() | |
| 43 | + | if err := persistToolResultState(nil, boundSession, result); err != nil { | |
| 44 | + | t.Fatal(err) | |
| 45 | + | } | |
| 51 | 46 | ||
| 52 | - | deadline := time.Now().Add(2 * time.Second) | |
| 53 | - | for transitionMu.TryLock() { | |
| 54 | - | transitionMu.Unlock() | |
| 55 | - | if time.Now().After(deadline) { | |
| 56 | - | persistMu.Unlock() | |
| 57 | - | t.Fatal("tool-result callback did not acquire the session read lock") | |
| 58 | - | } | |
| 59 | - | runtime.Gosched() | |
| 47 | + | if original.Meta.Goal == nil || original.Meta.Goal.Status != core.GoalDone { | |
| 48 | + | t.Fatalf("original goal = %#v, want done", original.Meta.Goal) | |
| 49 | + | } | |
| 50 | + | if currentSession.Meta.Goal == nil || currentSession.Meta.Goal.Status != core.GoalActive { | |
| 51 | + | t.Fatalf("current goal = %#v, want active", currentSession.Meta.Goal) | |
| 60 | 52 | } | |
| 53 | + | } | |
| 61 | 54 | ||
| 62 | - | transitioned := make(chan struct{}) | |
| 63 | - | go func() { | |
| 64 | - | transitionMu.Lock() | |
| 65 | - | persistMu.Lock() | |
| 66 | - | current = next | |
| 67 | - | persistMu.Unlock() | |
| 68 | - | transitionMu.Unlock() | |
| 69 | - | close(transitioned) | |
| 70 | - | }() | |
| 71 | - | persistMu.Unlock() | |
| 72 | - | select { | |
| 73 | - | case <-persisted: | |
| 74 | - | case <-time.After(2 * time.Second): | |
| 75 | - | t.Fatal("tool result did not persist") | |
| 55 | + | func TestPersistGoalToolResultRejectsBlockedWithoutReason(t *testing.T) { | |
| 56 | + | sess, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test") | |
| 57 | + | if err != nil { | |
| 58 | + | t.Fatal(err) | |
| 76 | 59 | } | |
| 77 | - | select { | |
| 78 | - | case <-transitioned: | |
| 79 | - | case <-time.After(2 * time.Second): | |
| 80 | - | t.Fatal("session transition did not finish") | |
| 60 | + | defer sess.Close() | |
| 61 | + | if err := sess.UpdateGoal(&core.SessionGoal{Objective: "finish the change", Status: core.GoalActive}); err != nil { | |
| 62 | + | t.Fatal(err) | |
| 81 | 63 | } | |
| 82 | 64 | ||
| 83 | - | if original.Meta.Goal == nil || original.Meta.Goal.Status != core.GoalDone { | |
| 84 | - | t.Fatalf("original goal = %#v, want done", original.Meta.Goal) | |
| 65 | + | result := core.ToolResult{Details: tools.GoalUpdate{Status: core.GoalBlocked}} | |
| 66 | + | if err := persistGoalToolResult(sess, result); err != nil { | |
| 67 | + | t.Fatal(err) | |
| 85 | 68 | } | |
| 86 | - | if next.Meta.Goal == nil || next.Meta.Goal.Status != core.GoalActive { | |
| 87 | - | t.Fatalf("next goal = %#v, want active", next.Meta.Goal) | |
| 69 | + | if got := sess.Meta.Goal; got == nil || got.Status != core.GoalActive { | |
| 70 | + | t.Fatalf("persisted goal = %#v, want active", got) | |
| 88 | 71 | } | |
| 89 | 72 | } | |
| 90 | 73 | ||
| @@ -7,7 +7,6 @@ import ( | |||||
| 7 | "fmt" | 7 | "fmt" | ||
| 8 | "os" | 8 | "os" | ||
| 9 | "path/filepath" | 9 | "path/filepath" | ||
| 10 | - | "runtime" | |||
| 11 | "strings" | 10 | "strings" | ||
| 12 | "sync" | 11 | "sync" | ||
| 13 | "testing" | 12 | "testing" | ||
| @@ -20,7 +19,7 @@ import ( | |||||
| 20 | "github.com/google/uuid" | 19 | "github.com/google/uuid" | ||
| 21 | ) | 20 | ) | ||
| 22 | 21 | ||||
| 23 | - | func TestPersistToolResultFinishesBeforeSessionTransition(t *testing.T) { | 22 | + | func TestToolResultPersistenceUsesBoundSession(t *testing.T) { |
| 24 | root := t.TempDir() | 23 | root := t.TempDir() | ||
| 25 | original, err := core.NewSession(root, "original", "provider", "model", "test") | 24 | original, err := core.NewSession(root, "original", "provider", "model", "test") | ||
| 26 | if err != nil { | 25 | if err != nil { | ||
| @@ -38,53 +37,37 @@ func TestPersistToolResultFinishesBeforeSessionTransition(t *testing.T) { | |||||
| 38 | } | 37 | } | ||
| 39 | } | 38 | } | ||
| 40 | 39 | ||||
| 41 | - | var transitionMu sync.RWMutex | 40 | + | boundSession := original |
| 42 | - | var persistMu sync.Mutex | 41 | + | currentSession := next // Simulate a completed session switch before callback delivery. |
| 43 | - | current := original | |||
| 44 | result := core.ToolResult{Details: tools.GoalUpdate{Status: core.GoalDone}} | 42 | result := core.ToolResult{Details: tools.GoalUpdate{Status: core.GoalDone}} | ||
| 45 | - | persistMu.Lock() | 43 | + | if err := persistToolResultState(nil, boundSession, result); err != nil { |
| 46 | - | persisted := make(chan struct{}) | 44 | + | t.Fatal(err) |
| 47 | - | go func() { | 45 | + | } |
| 48 | - | persistToolResultForCurrentSession(&transitionMu, &persistMu, func() *core.Session { return current }, nil, result) | |||
| 49 | - | close(persisted) | |||
| 50 | - | }() | |||
| 51 | 46 | ||||
| 52 | - | deadline := time.Now().Add(2 * time.Second) | 47 | + | if original.Meta.Goal == nil || original.Meta.Goal.Status != core.GoalDone { |
| 53 | - | for transitionMu.TryLock() { | 48 | + | t.Fatalf("original goal = %#v, want done", original.Meta.Goal) |
| 54 | - | transitionMu.Unlock() | 49 | + | } |
| 55 | - | if time.Now().After(deadline) { | 50 | + | if currentSession.Meta.Goal == nil || currentSession.Meta.Goal.Status != core.GoalActive { |
| 56 | - | persistMu.Unlock() | 51 | + | t.Fatalf("current goal = %#v, want active", currentSession.Meta.Goal) |
| 57 | - | t.Fatal("tool-result callback did not acquire the session read lock") | |||
| 58 | - | } | |||
| 59 | - | runtime.Gosched() | |||
| 60 | } | 52 | } | ||
| 53 | + | } | |||
| 61 | 54 | ||||
| 62 | - | transitioned := make(chan struct{}) | 55 | + | func TestPersistGoalToolResultRejectsBlockedWithoutReason(t *testing.T) { |
| 63 | - | go func() { | 56 | + | sess, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test") |
| 64 | - | transitionMu.Lock() | 57 | + | if err != nil { |
| 65 | - | persistMu.Lock() | 58 | + | t.Fatal(err) |
| 66 | - | current = next | |||
| 67 | - | persistMu.Unlock() | |||
| 68 | - | transitionMu.Unlock() | |||
| 69 | - | close(transitioned) | |||
| 70 | - | }() | |||
| 71 | - | persistMu.Unlock() | |||
| 72 | - | select { | |||
| 73 | - | case <-persisted: | |||
| 74 | - | case <-time.After(2 * time.Second): | |||
| 75 | - | t.Fatal("tool result did not persist") | |||
| 76 | } | 59 | } | ||
| 77 | - | select { | 60 | + | defer sess.Close() |
| 78 | - | case <-transitioned: | 61 | + | if err := sess.UpdateGoal(&core.SessionGoal{Objective: "finish the change", Status: core.GoalActive}); err != nil { |
| 79 | - | case <-time.After(2 * time.Second): | 62 | + | t.Fatal(err) |
| 80 | - | t.Fatal("session transition did not finish") | |||
| 81 | } | 63 | } | ||
| 82 | 64 | ||||
| 83 | - | if original.Meta.Goal == nil || original.Meta.Goal.Status != core.GoalDone { | 65 | + | result := core.ToolResult{Details: tools.GoalUpdate{Status: core.GoalBlocked}} |
| 84 | - | t.Fatalf("original goal = %#v, want done", original.Meta.Goal) | 66 | + | if err := persistGoalToolResult(sess, result); err != nil { |
| 67 | + | t.Fatal(err) | |||
| 85 | } | 68 | } | ||
| 86 | - | if next.Meta.Goal == nil || next.Meta.Goal.Status != core.GoalActive { | 69 | + | if got := sess.Meta.Goal; got == nil || got.Status != core.GoalActive { |
| 87 | - | t.Fatalf("next goal = %#v, want active", next.Meta.Goal) | 70 | + | t.Fatalf("persisted goal = %#v, want active", got) |
| 88 | } | 71 | } | ||
| 89 | } | 72 | } | ||
| 90 | 73 | ||||
M packages/agent/modes/goal.go +4 -0
| @@ -100,6 +100,10 @@ func (i *Interactive) startGoalContinuation(parent context.Context, message prov | |||
| 100 | 100 | pendingIdleWork := i.takePendingIdleWorkLocked() | |
| 101 | 101 | i.mu.Unlock() | |
| 102 | 102 | runPendingIdleWork(pendingIdleWork) | |
| 103 | + | // The objective may have been replaced while the prior turn was | |
| 104 | + | // finishing. Re-evaluate it now instead of waiting for another event; | |
| 105 | + | // the idle gate still gives queued user work priority. | |
| 106 | + | i.requestGoalContinuationIfIdle(parent) | |
| 103 | 107 | i.invalidate() | |
| 104 | 108 | return | |
| 105 | 109 | } | |
| @@ -100,6 +100,10 @@ func (i *Interactive) startGoalContinuation(parent context.Context, message prov | |||||
| 100 | pendingIdleWork := i.takePendingIdleWorkLocked() | 100 | pendingIdleWork := i.takePendingIdleWorkLocked() | ||
| 101 | i.mu.Unlock() | 101 | i.mu.Unlock() | ||
| 102 | runPendingIdleWork(pendingIdleWork) | 102 | runPendingIdleWork(pendingIdleWork) | ||
| 103 | + | // The objective may have been replaced while the prior turn was | |||
| 104 | + | // finishing. Re-evaluate it now instead of waiting for another event; | |||
| 105 | + | // the idle gate still gives queued user work priority. | |||
| 106 | + | i.requestGoalContinuationIfIdle(parent) | |||
| 103 | i.invalidate() | 107 | i.invalidate() | ||
| 104 | return | 108 | return | ||
| 105 | } | 109 | } | ||
M packages/agent/modes/goal_test.go +138 -6
| @@ -20,7 +20,7 @@ func goalToolRegistry() core.Registry { | |||
| 20 | 20 | type goalContinuationClient struct { | |
| 21 | 21 | mu sync.Mutex | |
| 22 | 22 | requests []provider.Request | |
| 23 | - | onSecond func() | |
| 23 | + | onCall func(call int) | |
| 24 | 24 | } | |
| 25 | 25 | ||
| 26 | 26 | func (c *goalContinuationClient) Name() string { return "goal-continuation" } | |
| @@ -29,10 +29,10 @@ func (c *goalContinuationClient) Stream(_ context.Context, req provider.Request) | |||
| 29 | 29 | c.mu.Lock() | |
| 30 | 30 | c.requests = append(c.requests, req) | |
| 31 | 31 | call := len(c.requests) | |
| 32 | - | onSecond := c.onSecond | |
| 32 | + | onCall := c.onCall | |
| 33 | 33 | c.mu.Unlock() | |
| 34 | - | if call == 2 && onSecond != nil { | |
| 35 | - | onSecond() | |
| 34 | + | if onCall != nil { | |
| 35 | + | onCall(call) | |
| 36 | 36 | } | |
| 37 | 37 | out := make(chan provider.Event, 1) | |
| 38 | 38 | out <- provider.EventDone{ | |
| @@ -60,8 +60,10 @@ func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) { | |||
| 60 | 60 | mu.Unlock() | |
| 61 | 61 | return nil | |
| 62 | 62 | } | |
| 63 | - | client := &goalContinuationClient{onSecond: func() { | |
| 64 | - | _ = persistGoal(&core.SessionGoal{Objective: "finish issue 97", Status: core.GoalPaused}) | |
| 63 | + | client := &goalContinuationClient{onCall: func(call int) { | |
| 64 | + | if call == 2 { | |
| 65 | + | _ = persistGoal(&core.SessionGoal{Objective: "finish issue 97", Status: core.GoalPaused}) | |
| 66 | + | } | |
| 65 | 67 | }} | |
| 66 | 68 | turnFlushed := make(chan struct{}, 2) | |
| 67 | 69 | interactive := NewInteractive(InteractiveConfig{ | |
| @@ -93,6 +95,136 @@ func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) { | |||
| 93 | 95 | } | |
| 94 | 96 | } | |
| 95 | 97 | ||
| 98 | + | func TestReplacementGoalContinuesAfterStaleTurn(t *testing.T) { | |
| 99 | + | var mu sync.Mutex | |
| 100 | + | goal := &core.SessionGoal{Objective: "replacement objective", Status: core.GoalActive} | |
| 101 | + | currentGoal := func() *core.SessionGoal { | |
| 102 | + | mu.Lock() | |
| 103 | + | defer mu.Unlock() | |
| 104 | + | return cloneSessionGoal(goal) | |
| 105 | + | } | |
| 106 | + | persistGoal := func(next *core.SessionGoal) error { | |
| 107 | + | mu.Lock() | |
| 108 | + | goal = cloneSessionGoal(next) | |
| 109 | + | mu.Unlock() | |
| 110 | + | return nil | |
| 111 | + | } | |
| 112 | + | client := &goalContinuationClient{onCall: func(call int) { | |
| 113 | + | if call == 1 { | |
| 114 | + | _ = persistGoal(&core.SessionGoal{Objective: "replacement objective", Status: core.GoalPaused}) | |
| 115 | + | } | |
| 116 | + | }} | |
| 117 | + | turnFlushed := make(chan struct{}, 1) | |
| 118 | + | interactive := NewInteractive(InteractiveConfig{ | |
| 119 | + | Agent: core.NewAgent(client, "model", "system", goalToolRegistry()), | |
| 120 | + | CurrentGoal: currentGoal, | |
| 121 | + | PersistGoal: persistGoal, | |
| 122 | + | FlushSession: func() { | |
| 123 | + | turnFlushed <- struct{}{} | |
| 124 | + | }, | |
| 125 | + | }) | |
| 126 | + | interactive.runCtx = context.Background() | |
| 127 | + | interactive.mu.Lock() | |
| 128 | + | interactive.busy = true | |
| 129 | + | interactive.mu.Unlock() | |
| 130 | + | stale := provider.Message{ | |
| 131 | + | Role: provider.RoleUser, | |
| 132 | + | Content: []provider.Content{provider.TextBlock{Text: "original objective"}}, | |
| 133 | + | Meta: map[string]string{goalContinueMetaKey: "true"}, | |
| 134 | + | } | |
| 135 | + | ||
| 136 | + | interactive.startGoalContinuation(context.Background(), stale) | |
| 137 | + | select { | |
| 138 | + | case <-turnFlushed: | |
| 139 | + | case <-time.After(2 * time.Second): | |
| 140 | + | t.Fatal("replacement goal did not start") | |
| 141 | + | } | |
| 142 | + | ||
| 143 | + | client.mu.Lock() | |
| 144 | + | defer client.mu.Unlock() | |
| 145 | + | if len(client.requests) != 1 { | |
| 146 | + | t.Fatalf("provider requests = %d, want 1", len(client.requests)) | |
| 147 | + | } | |
| 148 | + | tail := client.requests[0].Messages[len(client.requests[0].Messages)-1] | |
| 149 | + | if got := userMessageText(tail); !strings.Contains(got, "replacement objective") || strings.Contains(got, "original objective") { | |
| 150 | + | t.Fatalf("continuation message = %q", got) | |
| 151 | + | } | |
| 152 | + | } | |
| 153 | + | ||
| 154 | + | func TestToolResultPersistenceRunsOutsideTUILock(t *testing.T) { | |
| 155 | + | var transitionMu sync.RWMutex | |
| 156 | + | callbackStarted := make(chan struct{}) | |
| 157 | + | allowPersistence := make(chan struct{}) | |
| 158 | + | transitionHeld := make(chan struct{}) | |
| 159 | + | transitionDone := make(chan struct{}) | |
| 160 | + | handled := make(chan struct{}) | |
| 161 | + | agent := core.NewAgent(nil, "model", "system", goalToolRegistry()) | |
| 162 | + | agent.CommitToolResult = func(_ string, _ core.ToolResult) error { | |
| 163 | + | close(callbackStarted) | |
| 164 | + | <-allowPersistence | |
| 165 | + | transitionMu.RLock() | |
| 166 | + | transitionMu.RUnlock() | |
| 167 | + | return nil | |
| 168 | + | } | |
| 169 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |
| 170 | + | ||
| 171 | + | go func() { | |
| 172 | + | _ = agent.CommitToolResult("tool", core.ToolResult{}) | |
| 173 | + | interactive.handleEvent(core.EvToolResult{ID: "tool", Result: core.ToolResult{}}) | |
| 174 | + | close(handled) | |
| 175 | + | }() | |
| 176 | + | select { | |
| 177 | + | case <-callbackStarted: | |
| 178 | + | case <-time.After(2 * time.Second): | |
| 179 | + | t.Fatal("tool-result callback did not start") | |
| 180 | + | } | |
| 181 | + | go func() { | |
| 182 | + | transitionMu.Lock() | |
| 183 | + | close(transitionHeld) | |
| 184 | + | interactive.ApplySessionAgent(core.NewAgent(nil, "model", "system", goalToolRegistry()), "provider", "model") | |
| 185 | + | transitionMu.Unlock() | |
| 186 | + | close(transitionDone) | |
| 187 | + | }() | |
| 188 | + | select { | |
| 189 | + | case <-transitionHeld: | |
| 190 | + | case <-time.After(2 * time.Second): | |
| 191 | + | t.Fatal("session transition did not acquire its lock") | |
| 192 | + | } | |
| 193 | + | select { | |
| 194 | + | case <-transitionDone: | |
| 195 | + | case <-time.After(2 * time.Second): | |
| 196 | + | close(allowPersistence) | |
| 197 | + | t.Fatal("session transition deadlocked with tool-result persistence") | |
| 198 | + | } | |
| 199 | + | close(allowPersistence) | |
| 200 | + | select { | |
| 201 | + | case <-handled: | |
| 202 | + | case <-time.After(2 * time.Second): | |
| 203 | + | t.Fatal("tool-result callback did not finish") | |
| 204 | + | } | |
| 205 | + | } | |
| 206 | + | ||
| 207 | + | func TestGoalBadgeUsesDurableStatusAfterPersistenceFailure(t *testing.T) { | |
| 208 | + | goal := &core.SessionGoal{Objective: "finish issue 97", Status: core.GoalActive} | |
| 209 | + | interactive := NewInteractive(InteractiveConfig{ | |
| 210 | + | Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()), | |
| 211 | + | CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) }, | |
| 212 | + | }) | |
| 213 | + | // Core converts a goal update to an error result when durable persistence | |
| 214 | + | // fails, so the presentation layer must leave the active badge unchanged. | |
| 215 | + | result := core.ToolResult{ | |
| 216 | + | Content: []provider.Content{provider.TextBlock{Text: "tool result state could not be persisted"}}, | |
| 217 | + | IsError: true, | |
| 218 | + | } | |
| 219 | + | ||
| 220 | + | interactive.handleEvent(core.EvToolResult{ID: "goal", Result: result}) | |
| 221 | + | interactive.mu.Lock() | |
| 222 | + | defer interactive.mu.Unlock() | |
| 223 | + | if interactive.goalStatus != core.GoalActive { | |
| 224 | + | t.Fatalf("goal badge = %q, want durable active status", interactive.goalStatus) | |
| 225 | + | } | |
| 226 | + | } | |
| 227 | + | ||
| 96 | 228 | func TestGoalSlashCommandStartsAndControlsGoal(t *testing.T) { | |
| 97 | 229 | var goal *core.SessionGoal | |
| 98 | 230 | interactive := NewInteractive(InteractiveConfig{ | |
| @@ -20,7 +20,7 @@ func goalToolRegistry() core.Registry { | |||||
| 20 | type goalContinuationClient struct { | 20 | type goalContinuationClient struct { | ||
| 21 | mu sync.Mutex | 21 | mu sync.Mutex | ||
| 22 | requests []provider.Request | 22 | requests []provider.Request | ||
| 23 | - | onSecond func() | 23 | + | onCall func(call int) |
| 24 | } | 24 | } | ||
| 25 | 25 | ||||
| 26 | func (c *goalContinuationClient) Name() string { return "goal-continuation" } | 26 | func (c *goalContinuationClient) Name() string { return "goal-continuation" } | ||
| @@ -29,10 +29,10 @@ func (c *goalContinuationClient) Stream(_ context.Context, req provider.Request) | |||||
| 29 | c.mu.Lock() | 29 | c.mu.Lock() | ||
| 30 | c.requests = append(c.requests, req) | 30 | c.requests = append(c.requests, req) | ||
| 31 | call := len(c.requests) | 31 | call := len(c.requests) | ||
| 32 | - | onSecond := c.onSecond | 32 | + | onCall := c.onCall |
| 33 | c.mu.Unlock() | 33 | c.mu.Unlock() | ||
| 34 | - | if call == 2 && onSecond != nil { | 34 | + | if onCall != nil { |
| 35 | - | onSecond() | 35 | + | onCall(call) |
| 36 | } | 36 | } | ||
| 37 | out := make(chan provider.Event, 1) | 37 | out := make(chan provider.Event, 1) | ||
| 38 | out <- provider.EventDone{ | 38 | out <- provider.EventDone{ | ||
| @@ -60,8 +60,10 @@ func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) { | |||||
| 60 | mu.Unlock() | 60 | mu.Unlock() | ||
| 61 | return nil | 61 | return nil | ||
| 62 | } | 62 | } | ||
| 63 | - | client := &goalContinuationClient{onSecond: func() { | 63 | + | client := &goalContinuationClient{onCall: func(call int) { |
| 64 | - | _ = persistGoal(&core.SessionGoal{Objective: "finish issue 97", Status: core.GoalPaused}) | 64 | + | if call == 2 { |
| 65 | + | _ = persistGoal(&core.SessionGoal{Objective: "finish issue 97", Status: core.GoalPaused}) | |||
| 66 | + | } | |||
| 65 | }} | 67 | }} | ||
| 66 | turnFlushed := make(chan struct{}, 2) | 68 | turnFlushed := make(chan struct{}, 2) | ||
| 67 | interactive := NewInteractive(InteractiveConfig{ | 69 | interactive := NewInteractive(InteractiveConfig{ | ||
| @@ -93,6 +95,136 @@ func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) { | |||||
| 93 | } | 95 | } | ||
| 94 | } | 96 | } | ||
| 95 | 97 | ||||
| 98 | + | func TestReplacementGoalContinuesAfterStaleTurn(t *testing.T) { | |||
| 99 | + | var mu sync.Mutex | |||
| 100 | + | goal := &core.SessionGoal{Objective: "replacement objective", Status: core.GoalActive} | |||
| 101 | + | currentGoal := func() *core.SessionGoal { | |||
| 102 | + | mu.Lock() | |||
| 103 | + | defer mu.Unlock() | |||
| 104 | + | return cloneSessionGoal(goal) | |||
| 105 | + | } | |||
| 106 | + | persistGoal := func(next *core.SessionGoal) error { | |||
| 107 | + | mu.Lock() | |||
| 108 | + | goal = cloneSessionGoal(next) | |||
| 109 | + | mu.Unlock() | |||
| 110 | + | return nil | |||
| 111 | + | } | |||
| 112 | + | client := &goalContinuationClient{onCall: func(call int) { | |||
| 113 | + | if call == 1 { | |||
| 114 | + | _ = persistGoal(&core.SessionGoal{Objective: "replacement objective", Status: core.GoalPaused}) | |||
| 115 | + | } | |||
| 116 | + | }} | |||
| 117 | + | turnFlushed := make(chan struct{}, 1) | |||
| 118 | + | interactive := NewInteractive(InteractiveConfig{ | |||
| 119 | + | Agent: core.NewAgent(client, "model", "system", goalToolRegistry()), | |||
| 120 | + | CurrentGoal: currentGoal, | |||
| 121 | + | PersistGoal: persistGoal, | |||
| 122 | + | FlushSession: func() { | |||
| 123 | + | turnFlushed <- struct{}{} | |||
| 124 | + | }, | |||
| 125 | + | }) | |||
| 126 | + | interactive.runCtx = context.Background() | |||
| 127 | + | interactive.mu.Lock() | |||
| 128 | + | interactive.busy = true | |||
| 129 | + | interactive.mu.Unlock() | |||
| 130 | + | stale := provider.Message{ | |||
| 131 | + | Role: provider.RoleUser, | |||
| 132 | + | Content: []provider.Content{provider.TextBlock{Text: "original objective"}}, | |||
| 133 | + | Meta: map[string]string{goalContinueMetaKey: "true"}, | |||
| 134 | + | } | |||
| 135 | + | ||||
| 136 | + | interactive.startGoalContinuation(context.Background(), stale) | |||
| 137 | + | select { | |||
| 138 | + | case <-turnFlushed: | |||
| 139 | + | case <-time.After(2 * time.Second): | |||
| 140 | + | t.Fatal("replacement goal did not start") | |||
| 141 | + | } | |||
| 142 | + | ||||
| 143 | + | client.mu.Lock() | |||
| 144 | + | defer client.mu.Unlock() | |||
| 145 | + | if len(client.requests) != 1 { | |||
| 146 | + | t.Fatalf("provider requests = %d, want 1", len(client.requests)) | |||
| 147 | + | } | |||
| 148 | + | tail := client.requests[0].Messages[len(client.requests[0].Messages)-1] | |||
| 149 | + | if got := userMessageText(tail); !strings.Contains(got, "replacement objective") || strings.Contains(got, "original objective") { | |||
| 150 | + | t.Fatalf("continuation message = %q", got) | |||
| 151 | + | } | |||
| 152 | + | } | |||
| 153 | + | ||||
| 154 | + | func TestToolResultPersistenceRunsOutsideTUILock(t *testing.T) { | |||
| 155 | + | var transitionMu sync.RWMutex | |||
| 156 | + | callbackStarted := make(chan struct{}) | |||
| 157 | + | allowPersistence := make(chan struct{}) | |||
| 158 | + | transitionHeld := make(chan struct{}) | |||
| 159 | + | transitionDone := make(chan struct{}) | |||
| 160 | + | handled := make(chan struct{}) | |||
| 161 | + | agent := core.NewAgent(nil, "model", "system", goalToolRegistry()) | |||
| 162 | + | agent.CommitToolResult = func(_ string, _ core.ToolResult) error { | |||
| 163 | + | close(callbackStarted) | |||
| 164 | + | <-allowPersistence | |||
| 165 | + | transitionMu.RLock() | |||
| 166 | + | transitionMu.RUnlock() | |||
| 167 | + | return nil | |||
| 168 | + | } | |||
| 169 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |||
| 170 | + | ||||
| 171 | + | go func() { | |||
| 172 | + | _ = agent.CommitToolResult("tool", core.ToolResult{}) | |||
| 173 | + | interactive.handleEvent(core.EvToolResult{ID: "tool", Result: core.ToolResult{}}) | |||
| 174 | + | close(handled) | |||
| 175 | + | }() | |||
| 176 | + | select { | |||
| 177 | + | case <-callbackStarted: | |||
| 178 | + | case <-time.After(2 * time.Second): | |||
| 179 | + | t.Fatal("tool-result callback did not start") | |||
| 180 | + | } | |||
| 181 | + | go func() { | |||
| 182 | + | transitionMu.Lock() | |||
| 183 | + | close(transitionHeld) | |||
| 184 | + | interactive.ApplySessionAgent(core.NewAgent(nil, "model", "system", goalToolRegistry()), "provider", "model") | |||
| 185 | + | transitionMu.Unlock() | |||
| 186 | + | close(transitionDone) | |||
| 187 | + | }() | |||
| 188 | + | select { | |||
| 189 | + | case <-transitionHeld: | |||
| 190 | + | case <-time.After(2 * time.Second): | |||
| 191 | + | t.Fatal("session transition did not acquire its lock") | |||
| 192 | + | } | |||
| 193 | + | select { | |||
| 194 | + | case <-transitionDone: | |||
| 195 | + | case <-time.After(2 * time.Second): | |||
| 196 | + | close(allowPersistence) | |||
| 197 | + | t.Fatal("session transition deadlocked with tool-result persistence") | |||
| 198 | + | } | |||
| 199 | + | close(allowPersistence) | |||
| 200 | + | select { | |||
| 201 | + | case <-handled: | |||
| 202 | + | case <-time.After(2 * time.Second): | |||
| 203 | + | t.Fatal("tool-result callback did not finish") | |||
| 204 | + | } | |||
| 205 | + | } | |||
| 206 | + | ||||
| 207 | + | func TestGoalBadgeUsesDurableStatusAfterPersistenceFailure(t *testing.T) { | |||
| 208 | + | goal := &core.SessionGoal{Objective: "finish issue 97", Status: core.GoalActive} | |||
| 209 | + | interactive := NewInteractive(InteractiveConfig{ | |||
| 210 | + | Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()), | |||
| 211 | + | CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) }, | |||
| 212 | + | }) | |||
| 213 | + | // Core converts a goal update to an error result when durable persistence | |||
| 214 | + | // fails, so the presentation layer must leave the active badge unchanged. | |||
| 215 | + | result := core.ToolResult{ | |||
| 216 | + | Content: []provider.Content{provider.TextBlock{Text: "tool result state could not be persisted"}}, | |||
| 217 | + | IsError: true, | |||
| 218 | + | } | |||
| 219 | + | ||||
| 220 | + | interactive.handleEvent(core.EvToolResult{ID: "goal", Result: result}) | |||
| 221 | + | interactive.mu.Lock() | |||
| 222 | + | defer interactive.mu.Unlock() | |||
| 223 | + | if interactive.goalStatus != core.GoalActive { | |||
| 224 | + | t.Fatalf("goal badge = %q, want durable active status", interactive.goalStatus) | |||
| 225 | + | } | |||
| 226 | + | } | |||
| 227 | + | ||||
| 96 | func TestGoalSlashCommandStartsAndControlsGoal(t *testing.T) { | 228 | func TestGoalSlashCommandStartsAndControlsGoal(t *testing.T) { | ||
| 97 | var goal *core.SessionGoal | 229 | var goal *core.SessionGoal | ||
| 98 | interactive := NewInteractive(InteractiveConfig{ | 230 | interactive := NewInteractive(InteractiveConfig{ | ||
M packages/agent/modes/interactive.go +1 -5
| @@ -379,8 +379,7 @@ type InteractiveConfig struct { | |||
| 379 | 379 | CurrentGoal func() *core.SessionGoal | |
| 380 | 380 | PersistGoal func(goal *core.SessionGoal) error | |
| 381 | 381 | ||
| 382 | - | OnAssistant func(m provider.Message) | |
| 383 | - | OnToolResult func(id string, r core.ToolResult) | |
| 382 | + | OnAssistant func(m provider.Message) | |
| 384 | 383 | ||
| 385 | 384 | // Extensions, if non-nil, lets users invoke extension-registered | |
| 386 | 385 | // slash commands. Commands declared by extensions are looked up | |
| @@ -8301,9 +8300,6 @@ func (i *Interactive) handleEvent(ev core.AgentEvent) { | |||
| 8301 | 8300 | tc.Result = text.String() | |
| 8302 | 8301 | bumpToolRevision(tc) | |
| 8303 | 8302 | } | |
| 8304 | - | if i.cfg.OnToolResult != nil { | |
| 8305 | - | i.cfg.OnToolResult(e.ID, e.Result) | |
| 8306 | - | } | |
| 8307 | 8303 | if update, ok := tools.GoalUpdateFromResult(e.Result); ok && i.goalStatus == core.GoalActive { | |
| 8308 | 8304 | i.goalStatus = update.Status | |
| 8309 | 8305 | } | |
| @@ -379,8 +379,7 @@ type InteractiveConfig struct { | |||||
| 379 | CurrentGoal func() *core.SessionGoal | 379 | CurrentGoal func() *core.SessionGoal | ||
| 380 | PersistGoal func(goal *core.SessionGoal) error | 380 | PersistGoal func(goal *core.SessionGoal) error | ||
| 381 | 381 | ||||
| 382 | - | OnAssistant func(m provider.Message) | 382 | + | OnAssistant func(m provider.Message) |
| 383 | - | OnToolResult func(id string, r core.ToolResult) | |||
| 384 | 383 | ||||
| 385 | // Extensions, if non-nil, lets users invoke extension-registered | 384 | // Extensions, if non-nil, lets users invoke extension-registered | ||
| 386 | // slash commands. Commands declared by extensions are looked up | 385 | // slash commands. Commands declared by extensions are looked up | ||
| @@ -8301,9 +8300,6 @@ func (i *Interactive) handleEvent(ev core.AgentEvent) { | |||||
| 8301 | tc.Result = text.String() | 8300 | tc.Result = text.String() | ||
| 8302 | bumpToolRevision(tc) | 8301 | bumpToolRevision(tc) | ||
| 8303 | } | 8302 | } | ||
| 8304 | - | if i.cfg.OnToolResult != nil { | |||
| 8305 | - | i.cfg.OnToolResult(e.ID, e.Result) | |||
| 8306 | - | } | |||
| 8307 | if update, ok := tools.GoalUpdateFromResult(e.Result); ok && i.goalStatus == core.GoalActive { | 8303 | if update, ok := tools.GoalUpdateFromResult(e.Result); ok && i.goalStatus == core.GoalActive { | ||
| 8308 | i.goalStatus = update.Status | 8304 | i.goalStatus = update.Status | ||
| 8309 | } | 8305 | } | ||
M packages/agent/orchestration_modes.go +1 -1
| @@ -98,7 +98,7 @@ func runOrchestratedMode(parentCtx context.Context, args Args, version string, h | |||
| 98 | 98 | return err | |
| 99 | 99 | } | |
| 100 | 100 | r.Provider, r.Model = providerName, model | |
| 101 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | |
| 101 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } | |
| 102 | 102 | runtime.SetActiveSession(sess.ID) | |
| 103 | 103 | } | |
| 104 | 104 | announceSession(extMgr, sess) | |
| @@ -98,7 +98,7 @@ func runOrchestratedMode(parentCtx context.Context, args Args, version string, h | |||||
| 98 | return err | 98 | return err | ||
| 99 | } | 99 | } | ||
| 100 | r.Provider, r.Model = providerName, model | 100 | r.Provider, r.Model = providerName, model | ||
| 101 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | 101 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } |
| 102 | runtime.SetActiveSession(sess.ID) | 102 | runtime.SetActiveSession(sess.ID) | ||
| 103 | } | 103 | } | ||
| 104 | announceSession(extMgr, sess) | 104 | announceSession(extMgr, sess) | ||
M packages/agent/subagent_worker.go +1 -1
| @@ -143,7 +143,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||
| 143 | 143 | return err | |
| 144 | 144 | } | |
| 145 | 145 | r.Provider, r.Model = providerName, model | |
| 146 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | |
| 146 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } | |
| 147 | 147 | // A prompt is acknowledged by the supervisor only after this callback | |
| 148 | 148 | // has appended and synced the matching user message. This closes the | |
| 149 | 149 | // crash window between accepting a follow-up and writing session.json. | |
| @@ -143,7 +143,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||||
| 143 | return err | 143 | return err | ||
| 144 | } | 144 | } | ||
| 145 | r.Provider, r.Model = providerName, model | 145 | r.Provider, r.Model = providerName, model | ||
| 146 | - | ag.OnToolResult = func(_ string, result core.ToolResult) { persistExtensionToolResult(extMgr, sess, result) } | 146 | + | ag.CommitToolResult = func(_ string, result core.ToolResult) error { return persistToolResultState(extMgr, sess, result) } |
| 147 | // A prompt is acknowledged by the supervisor only after this callback | 147 | // A prompt is acknowledged by the supervisor only after this callback | ||
| 148 | // has appended and synced the matching user message. This closes the | 148 | // has appended and synced the matching user message. This closes the | ||
| 149 | // crash window between accepting a follow-up and writing session.json. | 149 | // crash window between accepting a follow-up and writing session.json. | ||
M packages/agent/tools/update_goal.go +3 -0
| @@ -88,6 +88,9 @@ func GoalUpdateFromResult(result core.ToolResult) (GoalUpdate, bool) { | |||
| 88 | 88 | if update.Status != core.GoalDone && update.Status != core.GoalBlocked { | |
| 89 | 89 | return GoalUpdate{}, false | |
| 90 | 90 | } | |
| 91 | + | if update.Status == core.GoalBlocked && strings.TrimSpace(update.Reason) == "" { | |
| 92 | + | return GoalUpdate{}, false | |
| 93 | + | } | |
| 91 | 94 | return update, true | |
| 92 | 95 | } | |
| 93 | 96 | ||
| @@ -88,6 +88,9 @@ func GoalUpdateFromResult(result core.ToolResult) (GoalUpdate, bool) { | |||||
| 88 | if update.Status != core.GoalDone && update.Status != core.GoalBlocked { | 88 | if update.Status != core.GoalDone && update.Status != core.GoalBlocked { | ||
| 89 | return GoalUpdate{}, false | 89 | return GoalUpdate{}, false | ||
| 90 | } | 90 | } | ||
| 91 | + | if update.Status == core.GoalBlocked && strings.TrimSpace(update.Reason) == "" { | |||
| 92 | + | return GoalUpdate{}, false | |||
| 93 | + | } | |||
| 91 | return update, true | 94 | return update, true | ||
| 92 | } | 95 | } | ||
| 93 | 96 | ||||
M packages/agent/tools/update_goal_test.go +17 -0
| @@ -41,6 +41,23 @@ func TestUpdateGoalToolRejectsUnknownFields(t *testing.T) { | |||
| 41 | 41 | } | |
| 42 | 42 | } | |
| 43 | 43 | ||
| 44 | + | func TestGoalUpdateFromResultRejectsBlockedWithoutReason(t *testing.T) { | |
| 45 | + | result := core.ToolResult{Details: GoalUpdate{Status: core.GoalBlocked}} | |
| 46 | + | if _, ok := GoalUpdateFromResult(result); ok { | |
| 47 | + | t.Fatal("blank-reason blocked update was accepted") | |
| 48 | + | } | |
| 49 | + | ||
| 50 | + | result.Details = GoalUpdate{Status: core.GoalBlocked, Reason: "needs user input"} | |
| 51 | + | if _, ok := GoalUpdateFromResult(result); !ok { | |
| 52 | + | t.Fatal("blocked update with reason was rejected") | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | result.Details = GoalUpdate{Status: core.GoalDone} | |
| 56 | + | if _, ok := GoalUpdateFromResult(result); !ok { | |
| 57 | + | t.Fatal("completed update without reason was rejected") | |
| 58 | + | } | |
| 59 | + | } | |
| 60 | + | ||
| 44 | 61 | func TestUpdateGoalToolRejectsUnsupportedStatus(t *testing.T) { | |
| 45 | 62 | tool := &UpdateGoalTool{} | |
| 46 | 63 | raw := json.RawMessage(`{"status":"active"}`) | |
| @@ -41,6 +41,23 @@ func TestUpdateGoalToolRejectsUnknownFields(t *testing.T) { | |||||
| 41 | } | 41 | } | ||
| 42 | } | 42 | } | ||
| 43 | 43 | ||||
| 44 | + | func TestGoalUpdateFromResultRejectsBlockedWithoutReason(t *testing.T) { | |||
| 45 | + | result := core.ToolResult{Details: GoalUpdate{Status: core.GoalBlocked}} | |||
| 46 | + | if _, ok := GoalUpdateFromResult(result); ok { | |||
| 47 | + | t.Fatal("blank-reason blocked update was accepted") | |||
| 48 | + | } | |||
| 49 | + | ||||
| 50 | + | result.Details = GoalUpdate{Status: core.GoalBlocked, Reason: "needs user input"} | |||
| 51 | + | if _, ok := GoalUpdateFromResult(result); !ok { | |||
| 52 | + | t.Fatal("blocked update with reason was rejected") | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | result.Details = GoalUpdate{Status: core.GoalDone} | |||
| 56 | + | if _, ok := GoalUpdateFromResult(result); !ok { | |||
| 57 | + | t.Fatal("completed update without reason was rejected") | |||
| 58 | + | } | |||
| 59 | + | } | |||
| 60 | + | ||||
| 44 | func TestUpdateGoalToolRejectsUnsupportedStatus(t *testing.T) { | 61 | func TestUpdateGoalToolRejectsUnsupportedStatus(t *testing.T) { | ||
| 45 | tool := &UpdateGoalTool{} | 62 | tool := &UpdateGoalTool{} | ||
| 46 | raw := json.RawMessage(`{"status":"active"}`) | 63 | raw := json.RawMessage(`{"status":"active"}`) | ||
M packages/core/agent.go +15 -4
| @@ -72,10 +72,12 @@ type Agent struct { | |||
| 72 | 72 | // collection into one lifecycle round trip. | |
| 73 | 73 | BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string) | |
| 74 | 74 | ||
| 75 | - | // OnToolResult, if set, receives extension/tool metadata immediately | |
| 76 | - | // after each tool call completes. Details are host-owned metadata and | |
| 77 | - | // are never sent to the provider; hosts may persist selected details | |
| 78 | - | // with the active session. | |
| 75 | + | // CommitToolResult, if set, commits extension/tool metadata immediately | |
| 76 | + | // after each tool call completes. Returning an error converts the result to | |
| 77 | + | // an error before it is appended to the transcript or sent to the provider. | |
| 78 | + | CommitToolResult func(id string, result ToolResult) error | |
| 79 | + | // OnToolResult observes the final result after CommitToolResult succeeds or | |
| 80 | + | // converts it to an error. | |
| 79 | 81 | OnToolResult func(id string, result ToolResult) | |
| 80 | 82 | ||
| 81 | 83 | // BeforeAssistantMessage, if set, is called after the model's | |
| @@ -861,6 +863,15 @@ func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink fun | |||
| 861 | 863 | continue | |
| 862 | 864 | } | |
| 863 | 865 | res := a.runOneTool(ctx, tc, tools, sink) | |
| 866 | + | if a.CommitToolResult != nil { | |
| 867 | + | if err := a.CommitToolResult(tc.ID, res); err != nil { | |
| 868 | + | res = ToolResult{ | |
| 869 | + | Content: []provider.Content{provider.TextBlock{Text: "tool result state could not be persisted"}}, | |
| 870 | + | IsError: true, | |
| 871 | + | Timing: res.Timing, | |
| 872 | + | } | |
| 873 | + | } | |
| 874 | + | } | |
| 864 | 875 | if res.IsError { | |
| 865 | 876 | hadError = true | |
| 866 | 877 | } | |
| @@ -72,10 +72,12 @@ type Agent struct { | |||||
| 72 | // collection into one lifecycle round trip. | 72 | // collection into one lifecycle round trip. | ||
| 73 | BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string) | 73 | BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string) | ||
| 74 | 74 | ||||
| 75 | - | // OnToolResult, if set, receives extension/tool metadata immediately | 75 | + | // CommitToolResult, if set, commits extension/tool metadata immediately |
| 76 | - | // after each tool call completes. Details are host-owned metadata and | 76 | + | // after each tool call completes. Returning an error converts the result to |
| 77 | - | // are never sent to the provider; hosts may persist selected details | 77 | + | // an error before it is appended to the transcript or sent to the provider. |
| 78 | - | // with the active session. | 78 | + | CommitToolResult func(id string, result ToolResult) error |
| 79 | + | // OnToolResult observes the final result after CommitToolResult succeeds or | |||
| 80 | + | // converts it to an error. | |||
| 79 | OnToolResult func(id string, result ToolResult) | 81 | OnToolResult func(id string, result ToolResult) | ||
| 80 | 82 | ||||
| 81 | // BeforeAssistantMessage, if set, is called after the model's | 83 | // BeforeAssistantMessage, if set, is called after the model's | ||
| @@ -861,6 +863,15 @@ func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink fun | |||||
| 861 | continue | 863 | continue | ||
| 862 | } | 864 | } | ||
| 863 | res := a.runOneTool(ctx, tc, tools, sink) | 865 | res := a.runOneTool(ctx, tc, tools, sink) | ||
| 866 | + | if a.CommitToolResult != nil { | |||
| 867 | + | if err := a.CommitToolResult(tc.ID, res); err != nil { | |||
| 868 | + | res = ToolResult{ | |||
| 869 | + | Content: []provider.Content{provider.TextBlock{Text: "tool result state could not be persisted"}}, | |||
| 870 | + | IsError: true, | |||
| 871 | + | Timing: res.Timing, | |||
| 872 | + | } | |||
| 873 | + | } | |||
| 874 | + | } | |||
| 864 | if res.IsError { | 875 | if res.IsError { | ||
| 865 | hadError = true | 876 | hadError = true | ||
| 866 | } | 877 | } | ||
M packages/core/intercept_test.go +38 -0
| @@ -42,6 +42,44 @@ func (r *recordingTool) Execute(_ context.Context, args json.RawMessage, _ func( | |||
| 42 | 42 | }, nil | |
| 43 | 43 | } | |
| 44 | 44 | ||
| 45 | + | func TestToolResultPersistenceFailureBecomesToolError(t *testing.T) { | |
| 46 | + | agent := NewAgent(nil, "test", "", Registry{ | |
| 47 | + | "result": &resultTool{result: ToolResult{ | |
| 48 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |
| 49 | + | Details: "durable update", | |
| 50 | + | }}, | |
| 51 | + | }) | |
| 52 | + | agent.CommitToolResult = func(string, ToolResult) error { return errors.New("disk full") } | |
| 53 | + | message := provider.Message{ | |
| 54 | + | Role: provider.RoleAssistant, | |
| 55 | + | Content: []provider.Content{provider.ToolCallBlock{ | |
| 56 | + | ID: "call-1", | |
| 57 | + | Name: "result", | |
| 58 | + | Arguments: json.RawMessage(`{}`), | |
| 59 | + | }}, | |
| 60 | + | } | |
| 61 | + | var event EvToolResult | |
| 62 | + | toolMessage, hadError := agent.executeTools(context.Background(), message, func(ev AgentEvent) { | |
| 63 | + | if result, ok := ev.(EvToolResult); ok { | |
| 64 | + | event = result | |
| 65 | + | } | |
| 66 | + | }) | |
| 67 | + | ||
| 68 | + | if !hadError { | |
| 69 | + | t.Fatal("persistence failure did not mark the tool batch failed") | |
| 70 | + | } | |
| 71 | + | if !event.Result.IsError || event.Result.Details != nil { | |
| 72 | + | t.Fatalf("tool event result = %#v", event.Result) | |
| 73 | + | } | |
| 74 | + | block, ok := toolMessage.Content[0].(provider.ToolResultBlock) | |
| 75 | + | if !ok || !block.IsError { | |
| 76 | + | t.Fatalf("tool transcript block = %#v", toolMessage.Content) | |
| 77 | + | } | |
| 78 | + | if got := block.Content[0].(provider.TextBlock).Text; got != "tool result state could not be persisted" { | |
| 79 | + | t.Fatalf("tool error text = %q", got) | |
| 80 | + | } | |
| 81 | + | } | |
| 82 | + | ||
| 45 | 83 | // TestBeforeToolExecuteModifiesArgs verifies that a non-nil | |
| 46 | 84 | // modifiedArgs returned from BeforeToolExecute is what the tool | |
| 47 | 85 | // actually sees. | |
| @@ -42,6 +42,44 @@ func (r *recordingTool) Execute(_ context.Context, args json.RawMessage, _ func( | |||||
| 42 | }, nil | 42 | }, nil | ||
| 43 | } | 43 | } | ||
| 44 | 44 | ||||
| 45 | + | func TestToolResultPersistenceFailureBecomesToolError(t *testing.T) { | |||
| 46 | + | agent := NewAgent(nil, "test", "", Registry{ | |||
| 47 | + | "result": &resultTool{result: ToolResult{ | |||
| 48 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |||
| 49 | + | Details: "durable update", | |||
| 50 | + | }}, | |||
| 51 | + | }) | |||
| 52 | + | agent.CommitToolResult = func(string, ToolResult) error { return errors.New("disk full") } | |||
| 53 | + | message := provider.Message{ | |||
| 54 | + | Role: provider.RoleAssistant, | |||
| 55 | + | Content: []provider.Content{provider.ToolCallBlock{ | |||
| 56 | + | ID: "call-1", | |||
| 57 | + | Name: "result", | |||
| 58 | + | Arguments: json.RawMessage(`{}`), | |||
| 59 | + | }}, | |||
| 60 | + | } | |||
| 61 | + | var event EvToolResult | |||
| 62 | + | toolMessage, hadError := agent.executeTools(context.Background(), message, func(ev AgentEvent) { | |||
| 63 | + | if result, ok := ev.(EvToolResult); ok { | |||
| 64 | + | event = result | |||
| 65 | + | } | |||
| 66 | + | }) | |||
| 67 | + | ||||
| 68 | + | if !hadError { | |||
| 69 | + | t.Fatal("persistence failure did not mark the tool batch failed") | |||
| 70 | + | } | |||
| 71 | + | if !event.Result.IsError || event.Result.Details != nil { | |||
| 72 | + | t.Fatalf("tool event result = %#v", event.Result) | |||
| 73 | + | } | |||
| 74 | + | block, ok := toolMessage.Content[0].(provider.ToolResultBlock) | |||
| 75 | + | if !ok || !block.IsError { | |||
| 76 | + | t.Fatalf("tool transcript block = %#v", toolMessage.Content) | |||
| 77 | + | } | |||
| 78 | + | if got := block.Content[0].(provider.TextBlock).Text; got != "tool result state could not be persisted" { | |||
| 79 | + | t.Fatalf("tool error text = %q", got) | |||
| 80 | + | } | |||
| 81 | + | } | |||
| 82 | + | ||||
| 45 | // TestBeforeToolExecuteModifiesArgs verifies that a non-nil | 83 | // TestBeforeToolExecuteModifiesArgs verifies that a non-nil | ||
| 46 | // modifiedArgs returned from BeforeToolExecute is what the tool | 84 | // modifiedArgs returned from BeforeToolExecute is what the tool | ||
| 47 | // actually sees. | 85 | // actually sees. | ||