diff --git a/internal/agent/runtime.go b/internal/agent/runtime.go index 496ac2a..6ac91be 100644 --- a/internal/agent/runtime.go +++ b/internal/agent/runtime.go @@ -2,8 +2,11 @@ package agent import ( "context" + "crypto/rand" + "encoding/hex" "errors" "fmt" + "log/slog" "strings" "time" @@ -76,8 +79,6 @@ type Turn struct { Steps int `json:"steps"` // Truncated is set when the run hit its step cap rather than finishing. Truncated bool `json:"truncated,omitempty"` - // History is the transcript to feed back into the next turn. - History []llm.Message `json:"-"` } // Run executes one turn against a garden, as actorID. @@ -100,14 +101,21 @@ func (r *Runner) Run(ctx context.Context, actorID, gardenID int64, message strin return nil, err } + // An id for this run, stamped on the change set so a row in the history list + // can be matched to the log lines that produced it. Without it, "the agent + // did something odd on Tuesday" has no thread back to what it was thinking. + runID := newRunID() + slog.Info("agent: run start", "run", runID, "garden", gardenID, "actor", actorID) + var ( result *agent.Result runErr error truncErr bool ) changeSet, err := r.svc.WithChangeSet(ctx, actorID, gardenID, service.ChangeSetOptions{ - Source: domain.SourceAgent, - Summary: turnSummary(message), + Source: domain.SourceAgent, + Summary: turnSummary(message), + AgentRunID: &runID, }, func(ctx context.Context) error { box := NewToolbox(r.svc, actorID) a := agent.New(r.model, systemPrompt(garden), @@ -143,7 +151,6 @@ func (r *Runner) Run(ctx context.Context, actorID, gardenID int64, message strin if result != nil { turn.Reply = result.Output turn.Steps = len(result.Steps) - turn.History = result.Messages } if turn.Reply == "" { turn.Reply = fallbackReply(turn) @@ -172,13 +179,29 @@ func fallbackReply(t *Turn) string { } } +// newRunID returns a short random identifier for one run. +func newRunID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err != nil { + // The id is for correlating logs, not for security. A clock-based + // fallback is worse than random and better than an empty string. + return fmt.Sprintf("t%d", time.Now().UnixNano()) + } + return hex.EncodeToString(b[:]) +} + // turnSummary is what the history list shows for this turn. The user's own words // are the most useful label available, trimmed to fit a list row. +// +// Trimmed by RUNES, not bytes: slicing a byte offset would cut a multibyte +// character in half and store invalid UTF-8 in the summary — which is not a +// hypothetical for text people type. func turnSummary(message string) string { const max = 120 s := strings.Join(strings.Fields(message), " ") - if len(s) > max { - s = strings.TrimSpace(s[:max]) + "…" + runes := []rune(s) + if len(runes) > max { + s = strings.TrimSpace(string(runes[:max])) + "…" } return s } diff --git a/internal/agent/runtime_test.go b/internal/agent/runtime_test.go index 91631f6..a331279 100644 --- a/internal/agent/runtime_test.go +++ b/internal/agent/runtime_test.go @@ -7,6 +7,8 @@ import ( "fmt" "strings" "testing" + "time" + "unicode/utf8" "gitea.stevedudenhoeffer.com/steve/majordomo/llm" "gitea.stevedudenhoeffer.com/steve/majordomo/provider/fake" @@ -277,3 +279,81 @@ func TestSystemPromptStatesTheCompassConvention(t *testing.T) { t.Error("system prompt doesn't state the garden's size") } } + +// TestPartialWorkSurvivesATimeout is the finding that mattered most on this PR. +// +// The run context carries a timeout. When it fires, WithChangeSet's recovery +// path has to record what already committed — and doing that with the SAME +// dead context would fail, losing the history for changes that really happened. +// The user-facing message says "anything I'd already changed is in History", so +// this isn't just a gap, it's a promise the code has to keep. +func TestPartialWorkSurvivesATimeout(t *testing.T) { + ctx := context.Background() + svc, owner := newAgentTestService(t) + g, err := svc.CreateGarden(ctx, owner, service.GardenInput{Name: "Plot", WidthCM: 2000, HeightCM: 2000}) + if err != nil { + t.Fatalf("garden: %v", err) + } + bed, err := svc.CreateObject(ctx, owner, g.ID, service.ObjectInput{ + Kind: domain.KindBed, Name: "Bed", XCM: 1000, YCM: 1000, WidthCM: 400, HeightCM: 400, + }) + if err != nil { + t.Fatalf("bed: %v", err) + } + before, _, _ := svc.GardenHistory(ctx, owner, g.ID, 0, 0) + + // A turn that renames the bed, then dies with the context already cancelled. + cancelled, cancel := context.WithCancel(ctx) + r := scriptedRunner(t, svc, + toolCall("move_object", map[string]any{ + "objectId": bed.ID, "xCm": 600.0, "yCm": 600.0, "version": bed.Version, + }), + fake.Step{Err: context.DeadlineExceeded}, + ) + // Cancel once the first tool call has landed, so the failure path runs with a + // dead context — exactly the timeout case. + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + _, err = r.Run(cancelled, owner, g.ID, "move the bed", nil, nil) + if err == nil { + t.Fatal("expected the turn to fail") + } + + // The move committed, so it must be in history and undoable. + after, _, herr := svc.GardenHistory(ctx, owner, g.ID, 0, 0) + if herr != nil { + t.Fatalf("history: %v", herr) + } + if len(after) != len(before)+1 { + t.Fatalf("the failed turn recorded %d change sets, want 1 — its work is otherwise un-undoable", + len(after)-len(before)) + } + if !strings.Contains(after[0].Summary, "failed partway") { + t.Errorf("summary = %q, want it marked as partial", after[0].Summary) + } + if _, conflicts, rerr := svc.RevertChangeSet(ctx, owner, after[0].ID, domain.SourceUI); rerr != nil || len(conflicts) != 0 { + t.Fatalf("the partial turn should be undoable: err=%v conflicts=%+v", rerr, conflicts) + } + o, _ := svc.DescribeGarden(ctx, owner, g.ID) + if len(o.Objects) > 0 && o.Objects[0].XCM != bed.XCM { + t.Errorf("undo left the bed at %v, want %v", o.Objects[0].XCM, bed.XCM) + } +} + +// TestTurnSummaryTrimsByRunes — slicing a byte offset would cut a multibyte +// character in half and store invalid UTF-8 in the history summary. +func TestTurnSummaryTrimsByRunes(t *testing.T) { + // 200 multibyte runes: a byte slice at 120 would land mid-character. + got := turnSummary(strings.Repeat("🌱", 200)) + if !utf8.ValidString(got) { + t.Errorf("turnSummary produced invalid UTF-8: %q", got) + } + if !strings.HasSuffix(got, "…") { + t.Errorf("long summary should be elided, got %q", got) + } + if n := utf8.RuneCountInString(got); n > 121 { + t.Errorf("summary is %d runes, want it trimmed", n) + } +} diff --git a/internal/api/agent.go b/internal/api/agent.go index 6e6b06c..2b19313 100644 --- a/internal/api/agent.go +++ b/internal/api/agent.go @@ -38,6 +38,9 @@ type chatEvent struct { Done *agent.Turn `json:"done,omitempty"` // Error is a turn that failed, in words meant for a person. Error string `json:"error,omitempty"` + // Warning rides alongside Done: the turn worked, but something adjacent to it + // didn't, and saying nothing would be the quieter lie. + Warning string `json:"warning,omitempty"` } type stepEvent struct { @@ -59,23 +62,7 @@ func (h *handlers) agentChat(c *gin.Context) { return } - // Headers before the first write, and a flush straight away: a proxy that - // buffers the response would reintroduce exactly the silence streaming is - // here to remove. - c.Header("Content-Type", "text/event-stream") - c.Header("Cache-Control", "no-cache") - c.Header("X-Accel-Buffering", "no") - c.Writer.Flush() - - send := func(ev chatEvent) { - b, err := json.Marshal(ev) - if err != nil { - slog.Error("api: encode chat event", "error", err) - return - } - _, _ = fmt.Fprintf(c.Writer, "data: %s\n\n", b) - c.Writer.Flush() - } + send := openEventStream(c) turn, err := h.agent.Run(c.Request.Context(), actor.ID, req.GardenID, req.Message, replayHistory(history), @@ -89,16 +76,45 @@ func (h *handlers) agentChat(c *gin.Context) { return } - // Record before announcing: if persistence fails, the user should see that - // their turn wasn't saved rather than a clean "done" followed by a thread - // that has forgotten it. - if _, err := h.svc.RecordAgentExchange(c.Request.Context(), actor.ID, req.GardenID, + // The turn itself succeeded — the garden really did change — so Done goes out + // regardless. But if the transcript couldn't be saved, say so: a clean "done" + // followed by a conversation that has forgotten the exchange after a reload is + // exactly the kind of quiet inconsistency that makes a tool feel unreliable. + // + // Detached from the request context, because the commonest reason this fails + // is the client having gone away — and the exchange is worth keeping either + // way, since the change set it produced certainly is. + if _, err := h.svc.RecordAgentExchange(context.WithoutCancel(c.Request.Context()), actor.ID, req.GardenID, req.Message, turn.Reply, turn.ChangeSetID); err != nil { slog.Error("api: record agent exchange", "error", err, "garden", req.GardenID) + send(chatEvent{Done: turn, Warning: "I couldn't save this exchange, so it won't be here after a reload. Anything I changed is still on the canvas, and in History."}) + return } send(chatEvent{Done: turn}) } +// openEventStream puts the response into SSE mode and returns a sender. +// +// Headers go out before the first write and the stream is flushed immediately, +// so a proxy holding the response until it looks complete can't reintroduce +// exactly the silence streaming exists to remove. +func openEventStream(c *gin.Context) func(chatEvent) { + c.Header("Content-Type", "text/event-stream") + c.Header("Cache-Control", "no-cache") + c.Header("X-Accel-Buffering", "no") + c.Writer.Flush() + + return func(ev chatEvent) { + b, err := json.Marshal(ev) + if err != nil { + slog.Error("api: encode chat event", "error", err) + return + } + _, _ = fmt.Fprintf(c.Writer, "data: %s\n\n", b) + c.Writer.Flush() + } +} + // getAgentHistory returns the actor's thread for a garden. func (h *handlers) getAgentHistory(c *gin.Context) { gardenID, ok := parseIDParam(c, "id") diff --git a/internal/service/revisions.go b/internal/service/revisions.go index ee59555..fc373c9 100644 --- a/internal/service/revisions.go +++ b/internal/service/revisions.go @@ -99,7 +99,8 @@ type ChangeSetOptions struct { Source string // Summary is the one-line description shown in the history list. Summary string - // AgentRunID joins this change set back to the executus run that produced it. + // AgentRunID joins this change set back to the agent run that produced it, + // so a change in the history list can be correlated with the run's log lines. AgentRunID *string } @@ -128,7 +129,11 @@ func (s *Service) WithChangeSet(ctx context.Context, actorID, gardenID int64, op // which is exactly the situation undo exists for. Record what happened, // mark the summary, and still report the failure. sc.summary = partialSummary(sc.summary) - if _, cerr := s.commitScope(ctx, sc, nil); cerr != nil { + // Detached from cancellation on purpose. The commonest reason fn failed is + // that ctx was cancelled or timed out — and using that same dead context + // to record what committed would fail too, losing the history for changes + // that really happened. That is precisely the case this path exists for. + if _, cerr := s.commitScope(context.WithoutCancel(ctx), sc, nil); cerr != nil { slog.Error("service: partial turn could not be recorded", "error", cerr, "garden", gardenID) } return nil, err @@ -306,8 +311,10 @@ func (s *Service) RevertChangeSet(ctx context.Context, actorID, changeSetID int6 changes, conflict, err := s.applyInverse(ctx, r, applied) if err != nil { // Record what actually landed before surfacing the failure, so the - // partial revert is visible and undoable rather than orphaned. - if _, cerr := s.commitScope(ctx, sc, &target.ID); cerr != nil { + // partial revert is visible and undoable rather than orphaned. Detached + // from cancellation for the same reason as WithChangeSet's path: a + // timed-out context can't be used to write the record of what it did. + if _, cerr := s.commitScope(context.WithoutCancel(ctx), sc, &target.ID); cerr != nil { slog.Error("service: partial revert could not be recorded", "error", cerr, "changeSet", changeSetID) } return nil, nil, err