From 71697ca4f837813d5da0dc6084cd581b611498dd Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 05:33:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(loop):=20agent=20loop=EF=BC=888=20?= =?UTF-8?q?=E8=BD=AE=20+=20checkpoint=20+=20=E6=B5=81=E5=BC=8F=20+=20?= =?UTF-8?q?=E8=AE=B0=E5=BF=86=E6=B3=A8=E5=85=A5/=E6=8F=90=E5=8F=96=20+=20t?= =?UTF-8?q?race=20+=20=E5=B7=A5=E5=85=B7=E8=B6=85=E6=97=B6=E5=9B=9E?= =?UTF-8?q?=E5=96=82=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/runtime/loop/checkpoint.go | 46 ++++ internal/runtime/loop/loop.go | 126 ++++++++++ internal/runtime/loop/loop_test.go | 375 ++++++++++++++++++++++++++++ internal/runtime/loop/memory.go | 83 ++++++ internal/runtime/loop/mocks_test.go | 204 +++++++++++++++ internal/runtime/loop/options.go | 56 +++++ internal/runtime/loop/request.go | 81 ++++++ internal/runtime/loop/run.go | 75 ++++++ internal/runtime/loop/step.go | 64 +++++ internal/runtime/loop/stream.go | 57 +++++ internal/runtime/loop/tools.go | 91 +++++++ internal/runtime/loop/trace.go | 34 +++ 12 files changed, 1292 insertions(+) create mode 100644 internal/runtime/loop/checkpoint.go create mode 100644 internal/runtime/loop/loop.go create mode 100644 internal/runtime/loop/loop_test.go create mode 100644 internal/runtime/loop/memory.go create mode 100644 internal/runtime/loop/mocks_test.go create mode 100644 internal/runtime/loop/options.go create mode 100644 internal/runtime/loop/request.go create mode 100644 internal/runtime/loop/run.go create mode 100644 internal/runtime/loop/step.go create mode 100644 internal/runtime/loop/stream.go create mode 100644 internal/runtime/loop/tools.go create mode 100644 internal/runtime/loop/trace.go diff --git a/internal/runtime/loop/checkpoint.go b/internal/runtime/loop/checkpoint.go new file mode 100644 index 0000000..1a21fd7 --- /dev/null +++ b/internal/runtime/loop/checkpoint.go @@ -0,0 +1,46 @@ +package loop + +import ( + "context" + "encoding/json" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" +) + +// checkpointState is the durable snapshot persisted at every step. Carrying the +// message window and the step index lets a long-running task be resumed from the +// latest checkpoint without losing context (design §1 durable execution). +type checkpointState struct { + Step int `json:"step"` + Messages []model.Message `json:"messages"` + StopReason string `json:"stop_reason,omitempty"` +} + +// saveCheckpoint persists a snapshot at step. It never fails the run: a nil store +// is a no-op and a marshal/save error is logged. +func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, messages []model.Message, stopReason string) { + if l.checkpoint == nil { + return + } + state := checkpointState{ + Step: step, + Messages: messages, + StopReason: stopReason, + } + buf, err := json.Marshal(state) + if err != nil { + l.log.Warn("checkpoint marshal", "step", step, "error", err) + return + } + cp := &persist.Checkpoint{ + ID: fmt.Sprintf("%s:%d", sessionID, step), + AgentID: sessionID, + Step: step, + State: buf, + } + if err := l.checkpoint.Save(ctx, cp); err != nil { + l.log.Warn("checkpoint save", "step", step, "error", err) + } +} diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go new file mode 100644 index 0000000..3528ba6 --- /dev/null +++ b/internal/runtime/loop/loop.go @@ -0,0 +1,126 @@ +// Package loop implements the default agent execution engine of the Cyrene v2 +// runtime (DevDocs/design/04-运行时.md §1). It owns the loop that package agent +// deliberately leaves out: assemble context from memory, drive the model for up +// to MaxRounds steps, execute any requested tool calls under a per-invoke +// timeout, feed tool results back to the model, persist a checkpoint at every +// step, emit structured traces, deliver streamed deltas and run the +// memory-extraction hook. +// +// The loop is configured with injected dependencies rather than concrete wiring, +// so it stays leaf-testable with mocks (a mock adapter, mock guardrail and mock +// tool, plus real in-memory memory/persist stores and a buffered trace recorder). +package loop + +import ( + "errors" + "io" + "log/slog" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Constants that bound the loop. These are compile-time so no magic number +// appears in the run path. +const ( + // DefaultMaxRounds is the tool-loop upper bound used when Options.MaxRounds + // is non-positive. It matches the agent package's own default. + DefaultMaxRounds = 8 + + // StopReasonMaxRounds reports that the loop exhausted its round budget + // before reaching a final answer. + StopReasonMaxRounds = "max_rounds" +) + +// Sentinel errors returned by the Loop. Use errors.Is to test for them; they are +// returned directly or wrapped with %w so the underlying cause stays available. +var ( + // ErrNilAgent is returned by Run when no *agent.Agent was configured. + ErrNilAgent = errors.New("loop: nil agent") + + // ErrNoAdapter is returned by Run (streaming path) when the agent was built + // without a model Adapter. + ErrNoAdapter = errors.New("loop: no model adapter") +) + +// Result is the outcome of a completed Run. +type Result struct { + // Text is the final assistant text, or empty when the loop exhausted its + // budget without a final answer. + Text string + // Steps is the number of model steps executed. + Steps int + // StopReason classifies how the loop ended: an agent StopReason* value or + // StopReasonMaxRounds. + StopReason string + // ToolCallsExecuted is the number of tool calls the loop invoked, including + // those that failed. + ToolCallsExecuted int +} + +// Loop is the default agent execution engine. It is configured once via New and +// safe for concurrent Run calls as long as the injected dependencies (agent, +// dispatcher, stores, recorder, tracer) are themselves concurrency-safe, which +// they are by contract. Run keeps all loop state on the stack. +type Loop struct { + agent *agent.Agent + tools *tool.Dispatcher + memory *memory.Store + retriever *memory.Retriever + working *memory.WorkingMemory + extractor memory.Extractor + checkpoint *persist.Store + tracer *trace.Tracer + guards *guardrail.Pipeline + personaID string + maxRounds int + onStream func(delta string) + log *slog.Logger +} + +// New builds a Loop from opts, applying defaults. A non-positive MaxRounds +// selects DefaultMaxRounds; a nil Log is replaced by a discard logger tagged with +// source "runtime/loop". Every dependency may be nil and is tolerated until the +// step that needs it (mirroring the run-time nil-tolerance of the underlying +// packages); Run returns ErrNilAgent when no agent is configured. +func New(opts Options) *Loop { + maxRounds := opts.MaxRounds + if maxRounds <= 0 { + maxRounds = DefaultMaxRounds + } + return &Loop{ + agent: opts.Agent, + tools: opts.Tools, + memory: opts.Memory, + retriever: opts.Retriever, + working: opts.WorkingMemory, + extractor: opts.Extractor, + checkpoint: opts.Checkpoint, + tracer: opts.Tracer, + guards: opts.Guards, + personaID: opts.PersonaID, + maxRounds: maxRounds, + onStream: opts.OnStream, + log: sourcedLog(opts.Log), + } +} + +// MaxRounds returns the tool-loop upper bound configured for the loop. +func (l *Loop) MaxRounds() int { return l.maxRounds } + +// Agent returns the injected agent. +func (l *Loop) Agent() *agent.Agent { return l.agent } + +// sourcedLog returns a logger tagged with the module source "runtime/loop", +// replacing a nil logger with a discard one so the loop can always log safely. +func sourcedLog(l *slog.Logger) *slog.Logger { + if l == nil { + l = slog.New(slog.NewTextHandler(io.Discard, nil)) + } + return logger.WithSource(l, "runtime/loop") +} diff --git a/internal/runtime/loop/loop_test.go b/internal/runtime/loop/loop_test.go new file mode 100644 index 0000000..eaf4408 --- /dev/null +++ b/internal/runtime/loop/loop_test.go @@ -0,0 +1,375 @@ +package loop + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +func TestRun_NoTools_FinalText(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Model: "m", Content: "Hello!"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, MaxRounds: 8}) + + res, err := l.Run(context.Background(), "sess-1", []model.Message{model.TextMessage(model.RoleUser, "hi")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "Hello!" { + t.Fatalf("Text = %q, want Hello!", res.Text) + } + if res.Steps != 1 { + t.Fatalf("Steps = %d, want 1", res.Steps) + } + if res.StopReason != agent.StopReasonStop { + t.Fatalf("StopReason = %q, want %q", res.StopReason, agent.StopReasonStop) + } + if res.ToolCallsExecuted != 0 { + t.Fatalf("ToolCallsExecuted = %d, want 0", res.ToolCallsExecuted) + } +} + +func TestRun_ToolCallThenFinal_FeedsBack(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "Echoed!"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + l := New(Options{Agent: a, MaxRounds: 8}) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "echo this")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "Echoed!" { + t.Fatalf("Text = %q, want Echoed!", res.Text) + } + if res.Steps != 2 { + t.Fatalf("Steps = %d, want 2", res.Steps) + } + if res.StopReason != agent.StopReasonStop { + t.Fatalf("StopReason = %q, want stop", res.StopReason) + } + if res.ToolCallsExecuted != 1 { + t.Fatalf("ToolCallsExecuted = %d, want 1", res.ToolCallsExecuted) + } + if len(ad.reqs) < 2 { + t.Fatalf("adapter requests = %d, want >= 2", len(ad.reqs)) + } + last := ad.reqs[len(ad.reqs)-1] + got := last.Messages[len(last.Messages)-1] + if got.Role != model.RoleTool { + t.Fatalf("last message role = %q, want tool", got.Role) + } + if !strings.Contains(got.Content[0].Text, "hi") { + t.Fatalf("tool result text = %q, want to contain 'hi'", got.Content[0].Text) + } +} + +func TestRun_MaxRounds(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: call}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "x"}) + a := newTestAgent(t, ad, td) + l := New(Options{Agent: a, MaxRounds: 2}) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Steps != 2 { + t.Fatalf("Steps = %d, want 2", res.Steps) + } + if res.StopReason != StopReasonMaxRounds { + t.Fatalf("StopReason = %q, want %q", res.StopReason, StopReasonMaxRounds) + } + if res.ToolCallsExecuted != 2 { + t.Fatalf("ToolCallsExecuted = %d, want 2", res.ToolCallsExecuted) + } +} + +func TestRun_ToolError_FeedsBack(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"boom","args":{}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "recovered"}}, + }} + td := dispatcherWith(t, &mockTool{name: "boom", err: errors.New("kaboom")}) + a := newTestAgent(t, ad, td) + l := New(Options{Agent: a, MaxRounds: 8}) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "boom now")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "recovered" { + t.Fatalf("Text = %q, want recovered", res.Text) + } + if res.ToolCallsExecuted != 1 { + t.Fatalf("ToolCallsExecuted = %d, want 1", res.ToolCallsExecuted) + } + got := ad.reqs[len(ad.reqs)-1].Messages[len(ad.reqs[len(ad.reqs)-1].Messages)-1] + if !strings.Contains(got.Content[0].Text, "kaboom") { + t.Fatalf("tool result should carry the error, got %q", got.Content[0].Text) + } +} + +func TestRun_ToolTimeout_FeedsBack(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"slow","args":{}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "continued after timeout"}}, + }} + td := dispatcherWithTimeout(t, 30*time.Millisecond, &mockTool{name: "slow", delay: 500 * time.Millisecond}) + a := newTestAgent(t, ad, td) + l := New(Options{Agent: a, MaxRounds: 8}) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "slow it")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "continued after timeout" { + t.Fatalf("Text = %q, want recovery text", res.Text) + } + got := ad.reqs[len(ad.reqs)-1].Messages[len(ad.reqs[len(ad.reqs)-1].Messages)-1] + if !strings.Contains(got.Content[0].Text, "timed out") { + t.Fatalf("tool result should report the timeout, got %q", got.Content[0].Text) + } +} + +func TestRun_CheckpointPerStep(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "done"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: a, Checkpoint: cp, MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil { + t.Fatalf("Run: %v", err) + } + got, err := cp.List(context.Background(), "sess") + if err != nil { + t.Fatalf("List: %v", err) + } + if len(got) != 2 { + t.Fatalf("checkpoints = %d, want 2", len(got)) + } + if got[0].Step != 1 || got[1].Step != 2 { + t.Fatalf("checkpoint steps = [%d,%d], want [1,2]", got[0].Step, got[1].Step) + } +} + +func TestRun_TraceSpans(t *testing.T) { + t.Parallel() + call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "done"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + var buf bytes.Buffer + tr := trace.New(trace.NewJSONRecorder(&buf)) + l := New(Options{Agent: a, Tracer: tr, MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil { + t.Fatalf("Run: %v", err) + } + saw := map[trace.SpanKind]bool{} + for _, r := range decodeSpanLines(t, buf.Bytes()) { + saw[r.Kind] = true + } + for _, k := range []trace.SpanKind{trace.SpanKindLLM, trace.SpanKindTool, trace.SpanKindStateTransition} { + if !saw[k] { + t.Fatalf("missing span kind %q in %+v", k, saw) + } + } +} + +func TestRun_MemoryInject(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + if _, err := store.Add(context.Background(), memory.MemoryItem{ + PersonaID: "persona-a", Content: "她喜欢喝冰美式", + Scope: memory.ScopePersona, Kind: memory.KindRelationship, + }); err != nil { + t.Fatalf("seed memory: %v", err) + } + ret := memory.NewRetriever(store) + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "喜欢冰美式"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, Retriever: ret, PersonaID: "persona-a", MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "sess", nil); err != nil { + t.Fatalf("Run: %v", err) + } + // An empty conversation still routes through injectMemory; the retriever + // returns the persona's items by recency when the query is blank. + if len(ad.reqs) == 0 { + t.Fatal("no request captured") + } + found := false + for _, m := range ad.reqs[0].Messages { + if m.Role == model.RoleSystem && strings.Contains(m.Content[0].Text, "冰美式") { + found = true + break + } + } + if !found { + t.Fatalf("injected memory not found in request messages: %+v", ad.reqs[0].Messages) + } +} + +func TestRun_MemoryExtract(t *testing.T) { + t.Parallel() + store := newTestMemoryStore(t) + ext := memory.NewManualExtractor("persona-a", "role-1") + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "记住了"}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, Memory: store, Extractor: ext, PersonaID: "persona-a", MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "记住 我爱喝冰美式")}); err != nil { + t.Fatalf("Run: %v", err) + } + items, err := store.ListByPersona(context.Background(), "persona-a") + if err != nil { + t.Fatalf("ListByPersona: %v", err) + } + if len(items) != 1 { + t.Fatalf("items = %d, want 1", len(items)) + } + if items[0].Content != "我爱喝冰美式" { + t.Fatalf("item content = %q, want 我爱喝冰美式", items[0].Content) + } +} + +func TestRun_StreamDeltas(t *testing.T) { + t.Parallel() + ad := &mockAdapter{ + stream: []mgResp{{resp: model.Response{Content: "Hello world"}}}, + chunks: [][]string{{"Hel", "lo ", "world"}}, + } + a := newTestAgent(t, ad, nil) + var mu sync.Mutex + var deltas []string + l := New(Options{ + Agent: a, + OnStream: func(d string) { mu.Lock(); deltas = append(deltas, d); mu.Unlock() }, + MaxRounds: 8, + }) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "Hello world" { + t.Fatalf("Text = %q, want Hello world", res.Text) + } + mu.Lock() + got := append([]string(nil), deltas...) + mu.Unlock() + if len(got) != 3 || got[0] != "Hel" || got[1] != "lo " || got[2] != "world" { + t.Fatalf("deltas = %+v, want [Hel lo world]", got) + } +} + +func TestRun_StreamGuardrailBlocks(t *testing.T) { + t.Parallel() + g := &mockGuard{inputVerdict: &guardrail.Verdict{Allow: false, Reason: "blocked in"}} + pipe := guardrail.New(guardrail.WithGuard(g)) + ad := &mockAdapter{} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, Guards: pipe, OnStream: func(string) {}, MaxRounds: 8}) + + res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.StopReason != agent.StopReasonGuardrailInput { + t.Fatalf("StopReason = %q, want %q", res.StopReason, agent.StopReasonGuardrailInput) + } + if res.Text != "blocked in" { + t.Fatalf("Text = %q, want blocked in", res.Text) + } +} + +func TestRun_ContextCanceled(t *testing.T) { + t.Parallel() + ad := &mockAdapter{} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, MaxRounds: 8}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")}); !errors.Is(err, context.Canceled) { + t.Fatalf("Run error = %v, want context.Canceled", err) + } +} + +func TestRun_NilAgent(t *testing.T) { + t.Parallel() + l := New(Options{}) + if _, err := l.Run(context.Background(), "sess", nil); !errors.Is(err, ErrNilAgent) { + t.Fatalf("Run error = %v, want ErrNilAgent", err) + } +} + +func TestNew_DefaultMaxRounds(t *testing.T) { + t.Parallel() + if got := New(Options{}).MaxRounds(); got != DefaultMaxRounds { + t.Fatalf("MaxRounds = %d, want %d", got, DefaultMaxRounds) + } + if got := New(Options{MaxRounds: 3}).MaxRounds(); got != 3 { + t.Fatalf("MaxRounds = %d, want 3", got) + } +} + +// decodeSpanLines reads every JSON span emitted as JSONL into recs, failing the +// test if any line is malformed. +func decodeSpanLines(t *testing.T, data []byte) []trace.SpanRecord { + t.Helper() + sc := bufio.NewScanner(bytes.NewReader(data)) + var recs []trace.SpanRecord + for sc.Scan() { + line := sc.Bytes() + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var r trace.SpanRecord + if err := json.Unmarshal(line, &r); err != nil { + t.Fatalf("decode span: %v", err) + } + recs = append(recs, r) + } + if err := sc.Err(); err != nil { + t.Fatalf("scan spans: %v", err) + } + return recs +} diff --git a/internal/runtime/loop/memory.go b/internal/runtime/loop/memory.go new file mode 100644 index 0000000..4eb817b --- /dev/null +++ b/internal/runtime/loop/memory.go @@ -0,0 +1,83 @@ +package loop + +import ( + "context" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// injectMemory retrieves the persona's relevant long-term memories and injects +// them into the first-turn context as a leading system message, so the model can +// ground its answer in recalled context (design §6 / §2). It returns the input +// messages unchanged when no retriever/persona is configured, retrieval fails, or +// nothing relevant is found. +func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []model.Message { + if l.retriever == nil || l.personaID == "" { + return messages + } + block, err := l.retriever.Inject(ctx, l.personaID, queryFromMessages(messages), 0) + if err != nil { + l.log.Warn("memory inject", "error", err) + return messages + } + if block == "" { + return messages + } + out := make([]model.Message, 0, len(messages)+1) + out = append(out, model.TextMessage(model.RoleSystem, block)) + out = append(out, messages...) + return out +} + +// extractMemory runs the extraction hook (design §6) over the current +// conversation and persists any derived items. Errors are logged, never fatal, so +// a bad extraction cannot abort the run. +func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step int) { + if l.extractor == nil || l.memory == nil { + return + } + items, err := l.extractor.Extract(ctx, messages) + if err != nil { + l.log.Warn("memory extract", "step", step, "error", err) + return + } + for _, it := range items { + if it.PersonaID == "" && l.personaID != "" { + it.PersonaID = l.personaID + } + if _, err := l.memory.Add(ctx, it); err != nil { + l.log.Warn("memory store", "step", step, "error", err) + } + } +} + +// queryFromMessages derives a retrieval query from the conversation: the last +// user turn's text, falling back to the last message's text, then "". +func queryFromMessages(messages []model.Message) string { + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role == model.RoleUser { + return messageText(messages[i]) + } + } + if n := len(messages); n > 0 { + return messageText(messages[n-1]) + } + return "" +} + +// messageText returns the concatenated plain-text payload of a message's text +// parts, which is the shared view used to build a retrieval query. +func messageText(m model.Message) string { + var b strings.Builder + for _, part := range m.Content { + if part.Type != model.ContentTypeText { + continue + } + if b.Len() > 0 { + b.WriteByte(' ') + } + b.WriteString(part.Text) + } + return b.String() +} diff --git a/internal/runtime/loop/mocks_test.go b/internal/runtime/loop/mocks_test.go new file mode 100644 index 0000000..15401ce --- /dev/null +++ b/internal/runtime/loop/mocks_test.go @@ -0,0 +1,204 @@ +package loop + +import ( + "context" + "io" + "log/slog" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" +) + +// mgResp is one scripted adapter result: a response plus an optional error. +type mgResp struct { + resp model.Response + err error +} + +// mockAdapter is a configurable model.Adapter for loop tests. It keeps two +// separate scripts and call counters (one for Chat, one for ChatStream) so a +// single instance serves both the non-streaming and streaming paths. Every +// request it is handed is recorded so tests can assert on the messages that were +// fed to the model (e.g. the injected memory or a fed-back tool result). +type mockAdapter struct { + name string + + chat []mgResp + chatCalls int + reqs []model.Request + + stream []mgResp + chunks [][]string + streamCalls int + streamReqs []model.Request +} + +func (m *mockAdapter) Name() string { + if m.name == "" { + return "mock" + } + return m.name +} + +func (m *mockAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + m.reqs = append(m.reqs, req) + if m.chatCalls >= len(m.chat) { + return model.Response{Content: "done"}, nil + } + r := m.chat[m.chatCalls] + m.chatCalls++ + return r.resp, r.err +} + +func (m *mockAdapter) ChatStream(_ context.Context, req model.Request, on func(string)) (model.Response, error) { + m.streamReqs = append(m.streamReqs, req) + i := m.streamCalls + m.streamCalls++ + if i < len(m.chunks) { + for _, c := range m.chunks[i] { + if on != nil { + on(c) + } + } + } + if i >= len(m.stream) { + return model.Response{Content: "done"}, nil + } + return m.stream[i].resp, m.stream[i].err +} + +// mockGuard is a configurable guardrail.Guardrail for loop tests. +type mockGuard struct { + inputVerdict *guardrail.Verdict + outputVerdict *guardrail.Verdict + inputErr error + outputErr error +} + +func (m *mockGuard) CheckInput(_ context.Context, _ *model.Request) (*guardrail.Verdict, error) { + return m.inputVerdict, m.inputErr +} + +func (m *mockGuard) CheckOutput(_ context.Context, _ *model.Response) (*guardrail.Verdict, error) { + return m.outputVerdict, m.outputErr +} + +// mockTool is a minimal sdk.Tool for loop tests. It supports a delay so a test +// can exercise the per-invoke timeout, and an error so a test can verify that a +// failed call is fed back to the model. +type mockTool struct { + name string + desc string + params map[string]any + result any + err error + delay time.Duration +} + +func (m *mockTool) Name() string { return m.name } + +func (m *mockTool) Description() string { + if m.desc == "" { + return "mock tool " + m.name + } + return m.desc +} + +func (m *mockTool) Parameters() (map[string]any, error) { return m.params, nil } + +func (m *mockTool) Execute(ctx context.Context, _ any) (any, error) { + if m.delay > 0 { + select { + case <-time.After(m.delay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return m.result, m.err +} + +// testLogger returns a logger that writes nowhere, keeping test output clean. +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// newTestAgent builds an Agent over ad with the given tools and no guardrails. +func newTestAgent(t *testing.T, ad model.Adapter, tools *tool.Dispatcher) *agent.Agent { + t.Helper() + a, err := agent.New("You are a helpful assistant.", tools, nil, agent.WithModel(ad, nil)) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + return a +} + +// dispatcherWith builds a Dispatcher with the default tool timeout. +func dispatcherWith(t *testing.T, tools ...sdk.Tool) *tool.Dispatcher { + t.Helper() + return dispatcherWithTimeout(t, 0, tools...) +} + +// dispatcherWithTimeout builds a Dispatcher applying a custom default timeout +// (a non-positive value selects the package default) and registering tools. +func dispatcherWithTimeout(t *testing.T, timeout time.Duration, tools ...sdk.Tool) *tool.Dispatcher { + t.Helper() + reg := registry.New() + var opts []tool.Option + if timeout > 0 { + opts = append(opts, tool.WithDefaultTimeout(timeout)) + } + d, err := tool.New(reg, opts...) + if err != nil { + t.Fatalf("tool.New: %v", err) + } + for _, tk := range tools { + if err := d.Register(tk); err != nil { + t.Fatalf("Register %q: %v", tk.Name(), err) + } + } + return d +} + +// newTestMemoryStore opens an isolated in-memory SQLite database and returns a +// memory.Store over it. Each call owns a private DB, so parallel tests are +// isolated by construction. +func newTestMemoryStore(t *testing.T) *memory.Store { + t.Helper() + ctx := context.Background() + d, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("db.Open: %v", err) + } + t.Cleanup(func() { _ = d.Close() }) + s, err := memory.NewStore(ctx, d, testLogger()) + if err != nil { + t.Fatalf("memory.NewStore: %v", err) + } + return s +} + +// newTestCheckpointStore opens an isolated in-memory SQLite database and returns +// a persist.Store over it. +func newTestCheckpointStore(t *testing.T) *persist.Store { + t.Helper() + ctx := context.Background() + d, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("db.Open: %v", err) + } + t.Cleanup(func() { _ = d.Close() }) + s, err := persist.New(ctx, d, testLogger()) + if err != nil { + t.Fatalf("persist.New: %v", err) + } + return s +} diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go new file mode 100644 index 0000000..4d853a9 --- /dev/null +++ b/internal/runtime/loop/options.go @@ -0,0 +1,56 @@ +package loop + +import ( + "log/slog" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Options configures a Loop. All fields are optional and nil is tolerated until +// the step that actually needs the field: the agent is required for Run (Run +// returns ErrNilAgent otherwise); a missing tool dispatcher causes a requested +// call to be fed back as an error; a missing retriever/persona skips memory +// injection; a missing extractor/store skips the memory hook; a missing +// checkpoint skips persistence; a missing tracer skips tracing. +type Options struct { + // Agent is the loop's execution unit: an instruction string plus a backend + // adapter, an optional tool dispatcher and optional guardrails. + Agent *agent.Agent + // Tools is the tool dispatcher used to execute model-requested calls. When + // nil the loop falls back to the agent's own dispatcher (if any). + Tools *tool.Dispatcher + // Memory is the long-term memory store used to persist extracted items. + Memory *memory.Store + // Retriever injects relevant memories into the first-turn context. + Retriever *memory.Retriever + // WorkingMemory is the per-session sliding window, provided for the + // orchestrator to inspect; the loop does not mutate it. + WorkingMemory *memory.WorkingMemory + // Extractor derives memory items to persist after every step. + Extractor memory.Extractor + // Checkpoint persists a durable snapshot of the loop state at every step. + Checkpoint *persist.Store + // Tracer emits a structured span for every step, LLM call and tool call. + Tracer *trace.Tracer + // Guards validates request/response payloads on the streaming path, where + // RunStep is bypassed. On the non-streaming path the agent runs its own + // guardrails inside RunStep. + Guards *guardrail.Pipeline + // PersonaID scopes memory retrieval and extraction. Blank skips both. + PersonaID string + // MaxRounds is the tool-loop upper bound. Non-positive selects + // DefaultMaxRounds (8). + MaxRounds int + // OnStream, when non-nil, switches the loop to the streaming path: the model + // adapter's ChatStream is driven directly and each token delta is forwarded + // to this callback. + OnStream func(delta string) + // Log is the base logger. Nil is replaced with a discard logger tagged with + // source "runtime/loop". + Log *slog.Logger +} diff --git a/internal/runtime/loop/request.go b/internal/runtime/loop/request.go new file mode 100644 index 0000000..08421ab --- /dev/null +++ b/internal/runtime/loop/request.go @@ -0,0 +1,81 @@ +package loop + +import ( + "encoding/json" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// buildRequest assembles a model.Request from the agent's public surface, +// mirroring agent.buildRequest so the streaming path behaves identically to +// RunStep (leading instruction message, tool schemas, model routing). +func (l *Loop) buildRequest(messages []model.Message) model.Request { + req := model.Request{ + Model: l.agent.Model(), + Messages: l.assembleMessages(messages), + Temperature: l.temperature(), + MaxTokens: l.agent.MaxTokens(), + Purpose: l.agent.Purpose(), + Tools: l.toolSchemas(), + } + if l.agent.Router() != nil { + if m, _ := l.agent.Router().RouteRequest(req); m != "" { + req.Model = m + } + } + return req +} + +// assembleMessages prepends the agent's instructions as a leading system message +// when they are non-empty, otherwise returns the input slice unchanged. +func (l *Loop) assembleMessages(messages []model.Message) []model.Message { + instr := l.agent.Instructions() + if instr == "" { + return messages + } + out := make([]model.Message, 0, len(messages)+1) + out = append(out, model.TextMessage(model.RoleSystem, instr)) + out = append(out, messages...) + return out +} + +// temperature returns a pointer to the configured temperature, mapping zero to +// nil so the provider default applies (matching the agent package). +func (l *Loop) temperature() *float64 { + t := l.agent.Temperature() + if t == 0 { + return nil + } + return &t +} + +// toolSchemas advertises the registered tool schemas to the model, or nil when no +// dispatcher is configured. +func (l *Loop) toolSchemas() []model.ToolSpec { + d := l.dispatcher() + if d == nil { + return nil + } + return d.ToolSchemas() +} + +// decodeToolCalls decodes tool calls from a model.Response using the same two +// conventions as the agent package: the normalized RoleTool placeholder, then a +// JSON envelope of the form {"tool_calls":[...]} in Content. It reports ok=true +// (with a non-empty slice) when a call is decoded. +func decodeToolCalls(resp model.Response) ([]agent.ToolCall, bool) { + if resp.Message.Role == model.RoleAssistant && resp.Message.ToolCallName != "" { + return []agent.ToolCall{{Name: resp.Message.ToolCallName, Args: map[string]any{}}}, true + } + var payload struct { + ToolCalls []agent.ToolCall `json:"tool_calls"` + } + if err := json.Unmarshal([]byte(resp.Content), &payload); err != nil { + return nil, false + } + if len(payload.ToolCalls) == 0 { + return nil, false + } + return payload.ToolCalls, true +} diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go new file mode 100644 index 0000000..487b5f3 --- /dev/null +++ b/internal/runtime/loop/run.go @@ -0,0 +1,75 @@ +package loop + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Run drives the agent loop: inject relevant memory into the first-turn context, +// then run up to MaxRounds steps. Each step runs the model (streaming through +// OnStream when configured) and any requested tool calls are executed under a +// per-invoke timeout and fed back as RoleTool messages so the model can continue. +// A checkpoint is persisted at every step, every step is traced as a +// state_transition span (with an llm_call child and tool_call spans for each +// invocation), and after each step the memory Extractor runs with errors logged +// rather than fatal. +// +// It returns the final answer, or a partial Result with StopReasonMaxRounds when +// the budget is exhausted without a final answer. Step errors are wrapped with %w +// so callers can classify them with errors.Is. +func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Message) (*Result, error) { + if ctx == nil { + ctx = context.Background() + } + if l.agent == nil { + return nil, fmt.Errorf("loop: run: %w", ErrNilAgent) + } + + msgs := l.injectMemory(ctx, messages) + res := &Result{} + for step := 1; step <= l.maxRounds; step++ { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("loop: run: step %d: %w", step, err) + } + + rc, roundSpan := l.span(ctx, "loop.step", trace.SpanKindStateTransition) + outcome, err := l.step(rc, sessionID, msgs) + if err != nil { + l.spanProp(roundSpan, "error", err.Error()) + l.finishSpan(roundSpan, trace.SpanStatusError) + return nil, fmt.Errorf("loop: run: step %d: %w", step, err) + } + l.spanProp(roundSpan, "step", step) + l.spanProp(roundSpan, "stop_reason", outcome.stopReason) + l.finishSpan(roundSpan, trace.SpanStatusOK) + + res.Steps = step + if outcome.final { + res.Text = outcome.text + res.StopReason = outcome.stopReason + l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason) + l.extractMemory(ctx, msgs, step) + return res, nil + } + + newMsgs, executed, err := l.runToolCalls(rc, msgs, outcome.toolCalls) + if err != nil { + return nil, fmt.Errorf("loop: run: step %d: tools: %w", step, err) + } + msgs = newMsgs + res.ToolCallsExecuted += executed + l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls) + l.extractMemory(ctx, msgs, step) + } + + // Budget exhausted without a final answer. + res.Steps = l.maxRounds + res.StopReason = StopReasonMaxRounds + l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds) + l.extractMemory(ctx, msgs, l.maxRounds) + return res, nil +} diff --git a/internal/runtime/loop/step.go b/internal/runtime/loop/step.go new file mode 100644 index 0000000..131ac1b --- /dev/null +++ b/internal/runtime/loop/step.go @@ -0,0 +1,64 @@ +package loop + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// stepOutcome is the loop's internal view of one model step, normalized from the +// agent's *agent.StepResult (non-streaming) or the equivalent built by the +// streaming path. +type stepOutcome struct { + final bool + text string + toolCalls []agent.ToolCall + stopReason string +} + +// step runs a single model call. When an OnStream callback is configured it +// drives the adapter's ChatStream directly, threading deltas to OnStream and +// applying the loop's guardrails; otherwise it delegates to the agent's RunStep, +// which runs the agent's own guardrails and records its own CallRecord. The whole +// call is wrapped in an llm_call span so the M9 panel sees each LLM invocation. +func (l *Loop) step(ctx context.Context, sessionID string, messages []model.Message) (*stepOutcome, error) { + _, llmSpan := l.span(ctx, "llm.call", trace.SpanKindLLM) + + var ( + out *stepOutcome + err error + ) + if l.onStream != nil { + out, err = l.streamStep(ctx, messages) + } else { + out, err = l.agentStep(ctx, sessionID, messages) + } + if err != nil { + l.spanProp(llmSpan, "error", err.Error()) + l.finishSpan(llmSpan, trace.SpanStatusError) + return nil, fmt.Errorf("loop: model step: %w", err) + } + + l.spanProp(llmSpan, "stop_reason", out.stopReason) + l.spanProp(llmSpan, "tool_calls", len(out.toolCalls)) + l.finishSpan(llmSpan, trace.SpanStatusOK) + return out, nil +} + +// agentStep delegates a non-streaming step to the agent, which runs input and +// output guardrails, calls the model and decodes any tool calls. +func (l *Loop) agentStep(ctx context.Context, sessionID string, messages []model.Message) (*stepOutcome, error) { + sr, err := l.agent.RunStep(ctx, sessionID, messages) + if err != nil { + return nil, fmt.Errorf("loop: agent step: %w", err) + } + return &stepOutcome{ + final: sr.Final, + text: sr.Text, + toolCalls: sr.ToolCalls, + stopReason: sr.StopReason, + }, nil +} diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go new file mode 100644 index 0000000..84fb20e --- /dev/null +++ b/internal/runtime/loop/stream.go @@ -0,0 +1,57 @@ +package loop + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// streamStep drives a streaming model call. It is the streaming counterpart of +// agent.RunStep: it builds the request from the agent's public surface, runs the +// loop's input guardrail, streams deltas through OnStream, runs the output +// guardrail and decodes any tool calls. Because RunStep does not expose a stream +// path, this method is owned by the loop; it mirrors the agent's semantics so the +// two behave identically apart from the delta callback. +func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepOutcome, error) { + adapter := l.agent.Adapter() + if adapter == nil { + return nil, fmt.Errorf("loop: stream step: %w", ErrNoAdapter) + } + req := l.buildRequest(messages) + + if l.guards != nil { + verdict, err := l.guards.CheckInput(ctx, &req) + if err != nil { + return nil, fmt.Errorf("loop: check input: %w", err) + } + if verdict != nil && !verdict.Allow { + return &stepOutcome{final: true, text: verdict.Reason, stopReason: agent.StopReasonGuardrailInput}, nil + } + } + + resp, err := adapter.ChatStream(ctx, req, func(delta string) { + if l.onStream != nil { + l.onStream(delta) + } + }) + if err != nil { + return nil, fmt.Errorf("loop: chat stream: %w", err) + } + + if l.guards != nil { + verdict, err := l.guards.CheckOutput(ctx, &resp) + if err != nil { + return nil, fmt.Errorf("loop: check output: %w", err) + } + if verdict != nil && !verdict.Allow { + return &stepOutcome{final: true, text: verdict.Reason, stopReason: agent.StopReasonGuardrailOutput}, nil + } + } + + if calls, ok := decodeToolCalls(resp); ok { + return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls}, nil + } + return &stepOutcome{final: true, text: resp.Content, stopReason: agent.StopReasonStop}, nil +} diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go new file mode 100644 index 0000000..b4b4cd6 --- /dev/null +++ b/internal/runtime/loop/tools.go @@ -0,0 +1,91 @@ +package loop + +import ( + "context" + "encoding/json" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// runToolCalls executes each requested tool call, appending a RoleTool result +// message (success or failure) so the model can continue. It returns the updated +// message slice and the number of tool calls invoked. Invocation is delegated to +// the dispatched tool, which applies the per-tool deadline; a missing dispatcher +// is fed back as an error so the loop never panics or crashes on an unhandled +// call. +func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls []agent.ToolCall) ([]model.Message, int, error) { + d := l.dispatcher() + out := messages + executed := 0 + for _, tc := range calls { + var ( + result any + invokeErr error + ) + if d == nil { + invokeErr = fmt.Errorf("no tool dispatcher configured for %q", tc.Name) + } else { + _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) + result, invokeErr = d.Invoke(ctx, tc.Name, tc.Args) + if invokeErr != nil { + l.spanProp(toolSpan, "error", invokeErr.Error()) + l.finishSpan(toolSpan, trace.SpanStatusError) + } else { + l.spanProp(toolSpan, "result", result) + l.finishSpan(toolSpan, trace.SpanStatusOK) + } + } + executed++ + out = append(out, toolResultMessage(tc, result, invokeErr)) + } + return out, executed, nil +} + +// dispatcher returns the loop's tool dispatcher, falling back to the agent's own +// when the loop was not given one. +func (l *Loop) dispatcher() *tool.Dispatcher { + if l.tools != nil { + return l.tools + } + if l.agent != nil { + return l.agent.Tools() + } + return nil +} + +// toolResultMessage renders a tool call's outcome as a RoleTool message ready to +// be fed back to the model. Errors are included in the text so the LLM can +// react to a failed or timed-out call. +func toolResultMessage(tc agent.ToolCall, result any, invokeErr error) model.Message { + text := fmt.Sprintf("tool %s result: %s", tc.Name, formatToolResult(result)) + if invokeErr != nil { + text = fmt.Sprintf("tool %s failed: %v", tc.Name, invokeErr) + } + return model.Message{Role: model.RoleTool, ToolCallName: tc.Name, + Content: []model.Content{{Type: model.ContentTypeText, Text: text}}} +} + +// formatToolResult renders a tool return value as text for the model. Strings are +// passed through (an empty string becomes "empty"); everything else is JSON +// encoded, falling back to a plain %v render on an encoding error. +func formatToolResult(result any) string { + switch v := result.(type) { + case nil: + return "null" + case string: + if v == "" { + return "empty" + } + return v + default: + b, err := json.Marshal(v) + if err != nil { + return fmt.Sprintf("%v", v) + } + return string(b) + } +} diff --git a/internal/runtime/loop/trace.go b/internal/runtime/loop/trace.go new file mode 100644 index 0000000..2f20881 --- /dev/null +++ b/internal/runtime/loop/trace.go @@ -0,0 +1,34 @@ +package loop + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// span starts a named span of the given kind when a Tracer is configured, else +// returns a nil no-op span. The returned context carries the span as the current +// one so nested spans chain as children. +func (l *Loop) span(ctx context.Context, name string, kind trace.SpanKind) (context.Context, *trace.Span) { + if l.tracer == nil { + return ctx, nil + } + return l.tracer.StartSpan(ctx, name, kind) +} + +// finishSpan closes a span with the given status, ignoring a nil span. +func (l *Loop) finishSpan(sp *trace.Span, status trace.SpanStatus) { + if sp == nil { + return + } + sp.Finish(status) +} + +// spanProp attaches a structured property to a span, ignoring a nil span and any +// mutation-after-close error. +func (l *Loop) spanProp(sp *trace.Span, key string, value any) { + if sp == nil { + return + } + _ = sp.SetProp(key, value) +}