feat(agent): integrate ReAct FSM into agent workflow

- Added a composite observer for transition events in the agent's Generate method.
- Implemented state transitions for various stages of the agent's processing, including preparation, execution, and error handling.
- Enhanced the processStepStream method to correctly track step indices and tool results.
- Updated the agent's context handling to improve observability and debugging.

go.mod

@@ -6,6 +6,8 @@ replace charm.land/fantasy v0.8.1 => ./internal/fantasy
 
 require (
 	charm.land/fantasy v0.8.1
+	github.com/ZanzyTHEbar/assert-lib v1.3.1
+	github.com/ZanzyTHEbar/errbuilder-go v1.5.1
 	github.com/adhocore/gronx v1.19.6
 	github.com/bwmarrin/discordgo v0.29.0
 	github.com/caarlos0/env/v11 v11.3.1
@@ -18,10 +20,12 @@ require (
 	github.com/openai/openai-go/v2 v2.7.1
 	github.com/pkoukk/tiktoken-go v0.1.6
 	github.com/pressly/goose/v3 v3.26.0
+	github.com/rs/zerolog v1.34.0
 	github.com/slack-go/slack v0.17.3
 	github.com/stretchr/testify v1.11.1
 	github.com/tencent-connect/botgo v0.2.1
 	github.com/tursodatabase/go-libsql v0.0.0-20251219133454-43644db490ff
+	golang.org/x/crypto v0.48.0
 	golang.org/x/oauth2 v0.35.0
 )
 
@@ -32,8 +36,6 @@ require (
 )
 
 require (
-	github.com/ZanzyTHEbar/assert-lib v1.3.1 // indirect
-	github.com/ZanzyTHEbar/errbuilder-go v1.5.1 // indirect
 	github.com/andybalholm/brotli v1.2.0 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/bytedance/gopkg v0.1.3 // indirect
@@ -61,7 +63,6 @@ require (
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mfridman/interpolate v0.0.2 // indirect
 	github.com/qmuntal/stateless v1.8.0 // indirect
-	github.com/rs/zerolog v1.34.0 // indirect
 	github.com/sethvargo/go-retry v0.3.0 // indirect
 	github.com/tidwall/gjson v1.18.0 // indirect
 	github.com/tidwall/match v1.2.0 // indirect
@@ -73,7 +74,6 @@ require (
 	github.com/valyala/fastjson v1.6.7 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/arch v0.24.0 // indirect
-	golang.org/x/crypto v0.48.0 // indirect
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
 	golang.org/x/net v0.50.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect

internal/fantasy/VENDORING.md

@@ -12,8 +12,7 @@ PicoClaw's `go.mod` contains a `replace` directive:
 replace charm.land/fantasy v0.8.1 => ./internal/fantasy
 ```
 
-This redirects all `charm.land/fantasy` imports to this local copy. No import paths
+This redirects all `charm.land/fantasy` imports to this local copy. No import paths need to change in either PicoClaw code or the fantasy source itself.
-need to change in either PicoClaw code or the fantasy source itself.
 
 ## Automated Sync System
 
@@ -69,14 +68,11 @@ The sync script will:
 - Update `.vendor-version`, `go.mod` replace directive, and this file
 - Run `go build` and `go test` for validation
 
-If any patch fails to apply, the script aborts with a clear error message showing
+If any patch fails to apply, the script aborts with a clear error message showing which patch conflicted. You'll need to resolve the conflict manually, then re-save the patch with `make fantasy-patch`.
-which patch conflicted. You'll need to resolve the conflict manually, then re-save
-the patch with `make fantasy-patch`.
 
 ## Patch Management
 
-Local modifications to the vendored SDK are tracked as numbered `.patch` files in the
+Local modifications to the vendored SDK are tracked as numbered `.patch` files in the `patches/` directory:
-`patches/` directory:
 
 ```
 internal/fantasy/patches/
@@ -119,7 +115,3 @@ If a patch fails during sync:
 | `patches/` | Directory of local modification patches |
 | `patches/.gitkeep` | Ensures the directory is tracked in git |
 | `VENDORING.md` | This documentation file |
-
-## Original License
-
-Fantasy is licensed under the MIT License. See `LICENSE` in this directory.

internal/fantasy/agent.go

@@ -377,6 +377,20 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 	var responseMessages []Message
 	var steps []StepResult
 
+	// Build a composite observer that fans out to all registered transition observers.
+	var fsmObserver ReActTransitionObserver
+	if len(a.settings.transitionObservers) > 0 {
+		obs := a.settings.transitionObservers
+		fsmObserver = ReActTransitionObserverFunc(func(ctx context.Context, t ReActTransition) {
+			for _, o := range obs {
+				o.OnReActTransition(ctx, t)
+			}
+		})
+	}
+	stepIdx := 0
+	fsm := newReActFSM(fsmObserver, &stepIdx)
+	fsm.Fire(ctx, ReActTriggerStart)
+
 	for {
 		stepInputMessages := append(initialPrompt, responseMessages...)
 		stepModel := a.settings.model
@@ -432,6 +446,8 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 			}
 		}
 
+		fsm.Fire(ctx, ReActTriggerPrepared)
+
 		preparedTools := a.prepareTools(stepTools, stepActiveTools, disableAllTools)
 
 		retryOptions := DefaultRetryOptions()
@@ -456,8 +472,10 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 			})
 		})
 		if err != nil {
+			fsm.Fire(ctx, ReActTriggerErrored)
 			return nil, err
 		}
+		fsm.Fire(ctx, ReActTriggerLLMResponded)
 
 		var stepToolCalls []ToolCallContent
 		for _, content := range result.Content {
@@ -472,6 +490,7 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 				stepToolCalls = append(stepToolCalls, validatedToolCall)
 			}
 		}
+		fsm.Fire(ctx, ReActTriggerToolsValidated)
 
 		var toolResults []ToolResultContent
 		if a.settings.toolRuntime != nil {
@@ -479,6 +498,7 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 		} else {
 			toolResults, err = a.executeTools(ctx, stepTools, stepToolCalls, nil)
 		}
+		fsm.Fire(ctx, ReActTriggerToolsExecuted)
 
 		// Build step content with validated tool calls and tool results
 		stepContent := []Content{}
@@ -503,6 +523,7 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 		}
 		currentStepMessages := toResponseMessages(stepContent)
 		responseMessages = append(responseMessages, currentStepMessages...)
+		fsm.Fire(ctx, ReActTriggerMessagesAppended)
 
 		stepResult := StepResult{
 			Response: Response{
@@ -515,6 +536,7 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 			Messages: currentStepMessages,
 		}
 		steps = append(steps, stepResult)
+		stepIdx = len(steps) - 1
 
 		for _, obs := range a.settings.stepObservers {
 			obs.OnReActStep(ctx, len(steps)-1, stepResult)
@@ -522,9 +544,15 @@ func (a *agent) Generate(ctx context.Context, opts AgentCall) (*AgentResult, err
 
 		shouldStop := isStopConditionMet(opts.StopWhen, steps)
 
-		if shouldStop || err != nil || len(stepToolCalls) == 0 || result.FinishReason != FinishReasonToolCalls {
+		if shouldStop {
+			fsm.Fire(ctx, ReActTriggerStopConditionMet)
 			break
 		}
+		if err != nil || len(stepToolCalls) == 0 || result.FinishReason != FinishReasonToolCalls {
+			fsm.Fire(ctx, ReActTriggerFinished)
+			break
+		}
+		fsm.Fire(ctx, ReActTriggerContinue)
 	}
 
 	totalUsage := Usage{}
@@ -1148,7 +1176,7 @@ func WithToolResultObserver(o ReActToolResultObserver) AgentOption {
 }
 
 // processStepStream processes a single step's stream and returns the step result.
-func (a *agent) processStepStream(ctx context.Context, stream StreamResponse, opts AgentStreamCall, _ []StepResult, stepTools []AgentTool) (stepExecutionResult, error) {
+func (a *agent) processStepStream(ctx context.Context, stream StreamResponse, opts AgentStreamCall, steps []StepResult, stepTools []AgentTool) (stepExecutionResult, error) {
 	var stepContent []Content
 	var stepToolCalls []ToolCallContent
 	var stepUsage Usage
@@ -1439,7 +1467,7 @@ func (a *agent) processStepStream(ctx context.Context, stream StreamResponse, op
 	for _, tr := range toolResults {
 		stepContent = append(stepContent, tr)
 		for _, obs := range a.settings.toolResultObservers {
-			obs.OnReActToolResult(ctx, 0, tr)
+			obs.OnReActToolResult(ctx, len(steps), tr)
 		}
 	}
 

internal/fantasy/go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/joho/godotenv v1.5.1
 	github.com/kaptinlin/jsonschema v0.6.10
 	github.com/openai/openai-go/v2 v2.7.1
+	github.com/qmuntal/stateless v1.8.0
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/oauth2 v0.35.0
 	google.golang.org/genai v1.45.0

internal/fantasy/go.sum

@@ -118,6 +118,8 @@ github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgm
 github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/qmuntal/stateless v1.8.0 h1:9+Eg/7bWLKxUxs/vysNYAelFAh85kTyueC3ee6v8im8=
+github.com/qmuntal/stateless v1.8.0/go.mod h1:KWa8KVzIBD/ZS0EdzL5oU79sGq7fKwH9WEFijTC5AWw=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=

internal/fantasy/tool_runtime_parallel.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"sync/atomic"
 )
 
 // ParallelToolRuntime executes tool calls concurrently when tools opt-in via
@@ -64,11 +65,11 @@ func (r ParallelToolRuntime) Execute(ctx context.Context, tools []AgentTool, too
 	}
 
 	sem := make(chan struct{}, maxConc)
-	inFlight := 0
+	var inFlight atomic.Int64
 	barrierWaits := 0
 	i := 0
 	emit := func() {
-		metrics(ToolRuntimeMetrics{Queued: len(toolCalls) - i, InFlightParallel: inFlight, BarrierWaits: barrierWaits})
+		metrics(ToolRuntimeMetrics{Queued: len(toolCalls) - i, InFlightParallel: int(inFlight.Load()), BarrierWaits: barrierWaits})
 	}
 	for i < len(toolCalls) {
 		if !isParallelSafe(toolCalls[i]) {
@@ -110,11 +111,11 @@ func (r ParallelToolRuntime) Execute(ctx context.Context, tools []AgentTool, too
 				defer wg.Done()
 				logEvent(ToolRuntimeLogEvent{Event: "dispatch", ToolCallID: tc.ToolCallID, ToolName: tc.ToolName})
 				sem <- struct{}{}
-				inFlight++
+				inFlight.Add(1)
 				emit()
 				defer func() {
 					<-sem
-					inFlight--
+					inFlight.Add(-1)
 					emit()
 				}()
 

pkg/agent/context.go

@@ -151,15 +151,18 @@ func (cb *ContextBuilder) BuildSystemPrompt() string {
 		parts = append(parts, bootstrapContent)
 	}
 
-	// Skills - show summary, AI can read full content with read_file tool
+	// Skills - show summary index and inline full definitions for direct use
 	skillsSummary := cb.skillsLoader.BuildSkillsSummary()
 	if skillsSummary != "" {
 		parts = append(parts, fmt.Sprintf(`# Skills
 
-The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
+The following skills extend your capabilities. Full definitions are included below.
 
 %s`, skillsSummary))
 	}
+	if skillsDefs := cb.loadSkills(); skillsDefs != "" {
+		parts = append(parts, skillsDefs)
+	}
 
 	// Observation block (stable prefix for prompt cache alignment)
 	if cb.observationBlock != "" {

pkg/agent/loop.go

@@ -64,6 +64,7 @@ type processOptions struct {
 	SessionKey      string // Session identifier for history/context
 	Channel         string // Target channel for tool execution
 	ChatID          string // Target chat ID for tool execution
+	SenderID        string // Originating sender identifier (for logging/audit)
 	UserMessage     string // User message content (may include prefix)
 	DefaultResponse string // Response when LLM returns empty
 	EnableSummary   bool   // Whether to trigger summarization
@@ -413,6 +414,7 @@ func (al *AgentLoop) ProcessDirectStreaming(ctx context.Context, content, sessio
 		SessionKey:      msg.SessionKey,
 		Channel:         msg.Channel,
 		ChatID:          msg.ChatID,
+		SenderID:        msg.SenderID,
 		UserMessage:     msg.Content,
 		DefaultResponse: "I've completed processing but have no response to give.",
 		EnableSummary:   true,
@@ -474,7 +476,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
 	})
 }
 
-func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
+func (al *AgentLoop) processSystemMessage(_ context.Context, msg bus.InboundMessage) (string, error) {
 	// Verify this is a system message
 	if msg.Channel != "system" {
 		return "", fmt.Errorf("processSystemMessage called with non-system message channel: %s", msg.Channel)
@@ -546,6 +548,12 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
 	}
 
 	// 1. Update tool contexts
+	logger.DebugCF("agent", "runAgentLoop: starting",
+		map[string]interface{}{
+			"session_key": opts.SessionKey,
+			"channel":     opts.Channel,
+			"sender_id":   opts.SenderID,
+		})
 	al.updateToolContexts(opts.Channel, opts.ChatID)
 
 	// 2. Load observation block for system prompt injection
@@ -598,6 +606,10 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
 	}
 
 	// 5. Convert history to Fantasy message format
+	logger.DebugCF("agent", "runAgentLoop: history messages",
+		map[string]interface{}{
+			"history": formatMessagesForLog(historyMsgs),
+		})
 	fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs)
 
 	// 6. Build adapted tools from PicoClaw registry (with optional offloading)
@@ -912,10 +924,19 @@ func (al *AgentLoop) updateToolContexts(channel, chatID string) {
 }
 
 // maybeSummarize triggers summarization if the session history exceeds thresholds.
+// At the critical threshold (≥95% of context window) it synchronously force-compresses
+// the history before the normal async summarization path runs.
 func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) {
 	newHistory := al.sessions.GetHistory(sessionKey)
 	tokenEstimate := al.estimateTokens(newHistory)
 	threshold := al.contextWindow * 75 / 100
+	criticalThreshold := al.contextWindow * 95 / 100
+
+	// Emergency path: drop oldest messages immediately when near context limit.
+	if tokenEstimate > criticalThreshold {
+		al.forceCompression(sessionKey)
+		return
+	}
 
 	if len(newHistory) > 20 || tokenEstimate > threshold {
 		if _, loading := al.summarizing.LoadOrStore(sessionKey, true); !loading {
@@ -1249,7 +1270,7 @@ func (al *AgentLoop) estimateTokens(msgs []messages.Message) int {
 	return totalChars * 2 / 5
 }
 
-func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) (string, bool) {
+func (al *AgentLoop) handleCommand(_ context.Context, msg bus.InboundMessage) (string, bool) {
 	content := strings.TrimSpace(msg.Content)
 	if !strings.HasPrefix(content, "/") {
 		return "", false

pkg/agent/loop_test.go

@@ -35,7 +35,9 @@ func (m *mockLanguageModel) Generate(_ context.Context, call fantasy.Call) (*fan
 
 func (m *mockLanguageModel) Stream(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
 	return func(yield func(fantasy.StreamPart) bool) {
-		yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, Delta: m.response})
+		if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, Delta: m.response}) {
+			return
+		}
 		yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop})
 	}, nil
 }

pkg/agent/tool_result_search.go

@@ -278,6 +278,9 @@ func loadView(ctx context.Context, kv KVDelegate, row sqlc.AgentToolResult, star
 	if sl < 1 {
 		sl = 1
 	}
+	if sl > len(lines) {
+		sl = len(lines)
+	}
 	if el > len(lines) {
 		el = len(lines)
 	}

pkg/fantasy/factory.go

@@ -20,6 +20,7 @@ import (
 
 // CreateProvider builds a Fantasy provider from PicoClaw config.
 // It mirrors the provider selection logic from the legacy providers.CreateProvider.
+// FIXME: we should use provider condigs and a handler, not hardcoded cases
 func CreateProvider(cfg *config.Config) (fantasy.Provider, error) {
 	model := cfg.Agents.Defaults.Model
 	providerName := strings.ToLower(cfg.Agents.Defaults.Provider)
@@ -67,14 +68,16 @@ func CreateProvider(cfg *config.Config) (fantasy.Provider, error) {
 
 // ModelID returns the effective model ID to pass to Fantasy's LanguageModel.
 // It strips provider prefixes that the old system used for routing.
+// FIXME: we should use provider condigs and a handler, not hardcoded cases
 func ModelID(cfg *config.Config) string {
 	model := cfg.Agents.Defaults.Model
 
 	// Strip provider prefix from model name (e.g., moonshot/kimi-k2.5 -> kimi-k2.5)
-	if idx := strings.Index(model, "/"); idx != -1 {
+	if before, after, ok := strings.Cut(model, "/"); ok {
-		prefix := model[:idx]
+		prefix := before
+		// FIXME: hardcoded provider prefixes are a hack
 		if prefix == "moonshot" || prefix == "nvidia" {
-			return model[idx+1:]
+			return after
 		}
 	}
 
@@ -82,6 +85,7 @@ func ModelID(cfg *config.Config) string {
 }
 
 // resolveProvider determines the API key, base URL, and proxy for a given config.
+// FIXME: we should use provider condigs and a handler, not hardcoded cases
 func resolveProvider(cfg *config.Config, providerName, model, lowerModel string) (apiKey, apiBase, proxy string) {
 	// First, try explicitly configured provider
 	if providerName != "" {
@@ -166,6 +170,7 @@ func resolveProvider(cfg *config.Config, providerName, model, lowerModel string)
 }
 
 // resolveProviderTimeout extracts the timeout from the matched provider config.
+// FIXME: we should use provider condigs and a handler, not hardcoded cases
 func resolveProviderTimeout(cfg *config.Config, providerName string) time.Duration {
 	var timeoutSec int
 

pkg/tools/registry_progressive_test.go

@@ -1,6 +1,7 @@
 package tools
 
 import (
+	"context"
 	"sort"
 	"testing"
 )
@@ -100,7 +101,7 @@ func TestProgressiveDisclosure_AllToolsStillDispatchable(t *testing.T) {
 
 	// tool_call should still dispatch to it
 	tc, _ := r.Get("tool_call")
-	result := tc.Execute(nil, map[string]interface{}{
+	result := tc.Execute(context.TODO(), map[string]interface{}{
 		"tool_name": "read_file",
 		"arguments": map[string]interface{}{},
 	})
@@ -118,7 +119,7 @@ func TestProgressiveDisclosure_SearchFindsHiddenTools(t *testing.T) {
 
 	// Even though read_file is hidden from Fantasy, tool_search should find it
 	ts, _ := r.Get("tool_search")
-	result := ts.Execute(nil, map[string]interface{}{"query": "read"})
+	result := ts.Execute(context.TODO(), map[string]interface{}{"query": "read"})
 
 	if result.IsError {
 		t.Fatalf("unexpected error: %s", result.ForLLM)

pkg/tools/retrieval.go

@@ -191,25 +191,25 @@ func formatSearchResults(source, query string, results []memory.SearchResult) st
 	}
 
 	var sb strings.Builder
-	sb.WriteString(fmt.Sprintf("Found %d results for '%s':\n\n", len(results), query))
+	fmt.Fprintf(&sb, "Found %d results for '%s':\n\n", len(results), query)
 
 	for i, r := range results {
-		sb.WriteString(fmt.Sprintf("%d. [%s] (score: %.2f) id=%s\n", i+1, r.Source, r.Score, r.ID))
+		fmt.Fprintf(&sb, "%d. [%s] (score: %.2f) id=%s\n", i+1, r.Source, r.Score, r.ID)
 
 		preview := r.Content
 		if len(preview) > 200 {
 			preview = preview[:200] + "..."
 		}
-		sb.WriteString(fmt.Sprintf("   %s\n", preview))
+		fmt.Fprintf(&sb, "   %s\n", preview)
 
 		if len(r.Metadata) > 0 {
 			var meta []string
 			for k, v := range r.Metadata {
 				meta = append(meta, fmt.Sprintf("%s=%s", k, v))
 			}
-			sb.WriteString(fmt.Sprintf("   meta: %s\n", strings.Join(meta, ", ")))
+			fmt.Fprintf(&sb, "   meta: %s\n", strings.Join(meta, ", "))
 		}
-		sb.WriteByte('\n')
+		fmt.Fprintf(&sb, "\n")
 	}
 
 	return sb.String()