Agent runtime: majordomo in-process, Ollama Cloud config, chat endpoint (#56) #70
@@ -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,6 +101,12 @@ 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
|
||||
@@ -108,6 +115,7 @@ func (r *Runner) Run(ctx context.Context, actorID, gardenID int64, message strin
|
||||
changeSet, err := r.svc.WithChangeSet(ctx, actorID, gardenID, service.ChangeSetOptions{
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
+37
-21
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user