Release/v0.0.7 (#16)

fix nats agent session
fix linter errors
add IngestOne for RAG
add Count for vectorstores
agent can return multiple handoffs

Author: henomis

a2a/client.go

@@ -211,7 +211,6 @@ func (c *Client) AsTool() (*llm.Tool, error) {
 	return llm.NewTool(agent.SanitizeToolName(c.card.Name), c.card.Description, handler)
 }
 
-
 // extractStatusMessage returns the first text part from a task status message,
 // or an empty string if the message is nil or contains no text parts.
 func extractStatusMessage(msg *sdka2a.Message) string {

a2a/message_test.go

@@ -21,8 +21,8 @@ import (
 
 	sdka2a "github.com/a2aproject/a2a-go/v2/a2a"
 
-	"github.com/henomis/phero/agent"
 	pheroA2A "github.com/henomis/phero/a2a"
+	"github.com/henomis/phero/agent"
 	"github.com/henomis/phero/llm"
 )
 

agent/agent.go

@@ -44,9 +44,10 @@ type Agent struct {
 // Result represents the final output of an agent after processing user input and executing any tool calls.
 type Result struct {
 	// Parts holds the multimodal content of the final assistant message.
-	Parts        []llm.ContentPart
-	HandoffAgent *Agent
-	Summary      *trace.RunSummary
+	Parts         []llm.ContentPart
+	// HandoffAgents lists the agents that were handed work; more than one means fan-out.
+	HandoffAgents []*Agent
+	Summary       *trace.RunSummary
 }
 
 // TextContent returns the concatenation of all text parts in the result.
@@ -185,7 +186,7 @@ func (a *Agent) Run(ctx context.Context, parts ...llm.ContentPart) (result *Resu
 	ctx = trace.WithTracer(ctx, a.tracer)
 	ctx = trace.WithAgentName(ctx, a.name)
 	stats := newRunStats(a.name)
-	handoffAgentName := ""
+	var handoffAgentNames []string
 
 	session, sessionIndex, err := a.prepareSession(ctx, parts, stats)
 	if err != nil {
@@ -218,7 +219,7 @@ func (a *Agent) Run(ctx context.Context, parts ...llm.ContentPart) (result *Resu
 			Timestamp:  time.Now(),
 		})
 
-		summary := stats.summary(iteration, handoffAgentName, err)
+		summary := stats.summary(iteration, handoffAgentNames, err)
 		if result != nil {
 			result.Summary = summary
 		}
@@ -249,13 +250,15 @@ func (a *Agent) Run(ctx context.Context, parts ...llm.ContentPart) (result *Resu
 		}
 
 		session = iterationResult.session
-		if iterationResult.handoffAgent != nil {
-			handoffAgentName = iterationResult.handoffAgent.Name()
+		if len(iterationResult.handoffAgents) > 0 {
+			for _, ha := range iterationResult.handoffAgents {
+				handoffAgentNames = append(handoffAgentNames, ha.Name())
+			}
 		}
 
 		// If finalMessage is nil, it means the agent executed tool calls and needs to call the LLM again.
 		if iterationResult.lastMessage != nil {
-			return &Result{Parts: iterationResult.lastMessage.Parts, HandoffAgent: iterationResult.handoffAgent}, nil
+			return &Result{Parts: iterationResult.lastMessage.Parts, HandoffAgents: iterationResult.handoffAgents}, nil
 		}
 	}
 }
@@ -307,9 +310,9 @@ func (a *Agent) saveSession(ctx context.Context, messages []llm.Message, session
 
 // agentIteration represents the result of one iteration of the agent loop.
 type agentIteration struct {
-	session      []llm.Message
-	lastMessage  *llm.Message
-	handoffAgent *Agent
+	session       []llm.Message
+	lastMessage   *llm.Message
+	handoffAgents []*Agent
 }
 
 // handleAgentIteration executes one iteration of the agent loop: it calls the LLM with the current messages,
@@ -345,26 +348,23 @@ func (a *Agent) handleAgentIteration(ctx context.Context, session []llm.Message,
 	}
 	wg.Wait()
 
-	// Append results in order; find the first handoff, if any.
-	var handoffAgent *Agent
+	// Append results in order; collect all handoffs (fan-out when more than one).
+	var handoffAgents []*Agent
+	var handoffMsg *llm.Message
 	for i, result := range results {
 		session = append(session, *result)
-		if hAgent, ok := a.handoffs[toolCalls[i].Function.Name]; ok && handoffAgent == nil {
-			handoffAgent = hAgent
+		if hAgent, ok := a.handoffs[toolCalls[i].Function.Name]; ok {
+			handoffAgents = append(handoffAgents, hAgent)
+			if handoffMsg == nil {
+				handoffMsg = result
+			}
 		}
 	}
 
-	if handoffAgent != nil {
-		// All tool results are preserved in session. Return the handoff tool's
+	if len(handoffAgents) > 0 {
+		// All tool results are preserved in session. Return the first handoff tool's
 		// result as lastMessage so callers receive it in Result.Parts.
-		var handoffMsg *llm.Message
-		for i, toolCall := range toolCalls {
-			if _, ok := a.handoffs[toolCall.Function.Name]; ok {
-				handoffMsg = results[i]
-				break
-			}
-		}
-		return agentIteration{session: session, lastMessage: handoffMsg, handoffAgent: handoffAgent}, nil
+		return agentIteration{session: session, lastMessage: handoffMsg, handoffAgents: handoffAgents}, nil
 	}
 
 	return agentIteration{session: session}, nil

agent/agent_test.go

@@ -262,8 +262,8 @@ func TestRun_Simple(t *testing.T) {
 	if result.TextContent() != "Hello, world!" {
 		t.Fatalf("expected %q, got %q", "Hello, world!", result.TextContent())
 	}
-	if result.HandoffAgent != nil {
-		t.Fatalf("expected no handoff agent, got %v", result.HandoffAgent)
+	if len(result.HandoffAgents) != 0 {
+		t.Fatalf("expected no handoff agents, got %v", result.HandoffAgents)
 	}
 }
 
@@ -436,11 +436,11 @@ func TestRun_Handoff(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Run: unexpected error: %v", err)
 	}
-	if result.HandoffAgent == nil {
-		t.Fatal("expected HandoffAgent to be set")
+	if len(result.HandoffAgents) == 0 {
+		t.Fatal("expected HandoffAgents to be set")
 	}
-	if result.HandoffAgent.Name() != "worker" {
-		t.Fatalf("expected handoff to %q, got %q", "worker", result.HandoffAgent.Name())
+	if result.HandoffAgents[0].Name() != "worker" {
+		t.Fatalf("expected handoff to %q, got %q", "worker", result.HandoffAgents[0].Name())
 	}
 }
 
@@ -561,8 +561,8 @@ func multiToolCallResult(calls ...llm.ToolCall) *llm.Result {
 
 func toolCall(toolName, callID, arguments string) llm.ToolCall {
 	return llm.ToolCall{
-		ID:   callID,
-		Type: llm.ToolTypeFunction,
+		ID:       callID,
+		Type:     llm.ToolTypeFunction,
 		Function: llm.FunctionCall{Name: toolName, Arguments: arguments},
 	}
 }
@@ -676,10 +676,10 @@ func TestRun_HandoffWithPrecedingToolCall(t *testing.T) {
 	if !regularInvoked {
 		t.Error("regular_tool was not invoked; expected it to run even when a handoff is in the same batch")
 	}
-	if result.HandoffAgent == nil {
-		t.Fatal("expected HandoffAgent to be set")
+	if len(result.HandoffAgents) == 0 {
+		t.Fatal("expected HandoffAgents to be set")
 	}
-	if result.HandoffAgent.Name() != "worker" {
-		t.Errorf("HandoffAgent = %q, want %q", result.HandoffAgent.Name(), "worker")
+	if result.HandoffAgents[0].Name() != "worker" {
+		t.Errorf("HandoffAgents[0] = %q, want %q", result.HandoffAgents[0].Name(), "worker")
 	}
 }

agent/trace.go

@@ -94,7 +94,7 @@ func (s *runStats) recordMemorySave(count int, duration time.Duration) {
 	s.memorySaved += count
 }
 
-func (s *runStats) summary(iterations int, handoffAgent string, err error) *trace.RunSummary {
+func (s *runStats) summary(iterations int, handoffAgents []string, err error) *trace.RunSummary {
 	tools := make([]trace.ToolCallSummary, 0, len(s.toolSummaries))
 	toolNames := make([]string, 0, len(s.toolSummaries))
 	for toolName := range s.toolSummaries {
@@ -129,7 +129,7 @@ func (s *runStats) summary(iterations int, handoffAgent string, err error) *trac
 			Memory: s.memoryDuration,
 		},
 		Tools:        tools,
-		HandoffAgent: handoffAgent,
+		HandoffAgents: handoffAgents,
 	}
 
 	if err != nil {

examples/handoff/main.go

@@ -168,9 +168,9 @@ func main() {
 				break
 			}
 
-			if result.HandoffAgent != nil {
-				fmt.Printf("[handoff] %s → %s\n", routingAgent.Name(), result.HandoffAgent.Name())
-				routingAgent = result.HandoffAgent
+			if len(result.HandoffAgents) > 0 {
+				fmt.Printf("[handoff] %s → %s\n", routingAgent.Name(), result.HandoffAgents[0].Name())
+				routingAgent = result.HandoffAgents[0]
 				// Empty input: the specialist reads context from shared memory.
 				currentInput = ""
 				continue

examples/nats-agent/server/main.go

@@ -30,6 +30,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/nats-io/nats.go"
 
 	"github.com/henomis/phero/agent"
@@ -66,7 +67,7 @@ func main() {
 
 	srv, err := natsagent.New(nc, a, *owner, *name,
 		natsagent.WithAgentID("phero"),
-		natsagent.WithSession(*name),
+		natsagent.WithSession(uuid.NewString()),
 		natsagent.WithHeartbeatInterval(10*time.Second),
 	)
 	if err != nil {

examples/rag-chatbot/main.go

@@ -123,7 +123,7 @@ func main() {
 		return
 	}
 
-	if err := ragEngine.Ingest(bootstrapCtx, splitter); err != nil {
+	if err := ragEngine.IngestOnce(bootstrapCtx, splitter); err != nil {
 		fmt.Fprintf(os.Stderr, "failed to ingest file: %v\n", err)
 		return
 	}

nats/errors.go

@@ -20,8 +20,8 @@ var (
 	// ErrNilConn is returned when a nil *nats.Conn is passed to New or NewClient.
 	ErrNilConn = errors.New("nats: NATS connection must not be nil")
 
-	// ErrNilAgent is returned when a nil *agent.Agent is passed to New.
-	ErrNilAgent = errors.New("nats: agent must not be nil")
+	// ErrNilHandler is returned when a nil Handler is passed to New.
+	ErrNilHandler = errors.New("nats: handler must not be nil")
 
 	// ErrEmptyOwner is returned when the owner field is empty.
 	ErrEmptyOwner = errors.New("nats: owner must not be empty")

nats/server.go

@@ -21,40 +21,47 @@ import (
 	"time"
 
 	"github.com/henomis/phero/agent"
+	"github.com/henomis/phero/llm"
 	natsclient "github.com/nats-io/nats.go"
 	natsio "github.com/nats-io/nats.go/micro"
 )
 
 const protocolVersion = "0.3"
 
-// Server registers a Phero agent as a NATS micro service implementing the
+// Handler is implemented by anything that can process a prompt — *agent.Agent
+// and workflow executors both satisfy it.
+type Handler interface {
+	Run(ctx context.Context, parts ...llm.ContentPart) (*agent.Result, error)
+}
+
+// Server registers a Handler as a NATS micro service implementing the
 // NATS Agent Protocol v0.3. It handles:
 //
 //   - Service registration and discovery via $SRV.PING/INFO.agents (§3, §4).
 //   - Streaming prompt responses on the prompt endpoint (§5, §6).
 //   - Heartbeat publication on agents.hb.{agent}.{owner}.{name} (§8).
 //   - On-demand status replies on the status endpoint (§8.7).
 type Server struct {
-	nc    *natsclient.Conn
-	agent *agent.Agent
-	cfg   *serverConfig
-	owner string
-	name  string
+	nc      *natsclient.Conn
+	handler Handler
+	cfg     *serverConfig
+	owner   string
+	name    string
 
 	svc natsio.Service
 	wg  sync.WaitGroup
 }
 
-// New creates a Server that wraps a ready to serve on NATS.
+// New creates a Server that wraps h and serves it on NATS.
 // owner and name are required positional arguments (§3.2):
 //   - owner identifies the operator or account.
 //   - name is the per-instance label (the 5th token in the subject hierarchy).
-func New(nc *natsclient.Conn, a *agent.Agent, owner, name string, opts ...ServerOption) (*Server, error) {
+func New(nc *natsclient.Conn, h Handler, owner, name string, opts ...ServerOption) (*Server, error) {
 	if nc == nil {
 		return nil, ErrNilConn
 	}
-	if a == nil {
-		return nil, ErrNilAgent
+	if h == nil {
+		return nil, ErrNilHandler
 	}
 	if owner == "" {
 		return nil, ErrEmptyOwner
@@ -70,12 +77,17 @@ func New(nc *natsclient.Conn, a *agent.Agent, owner, name string, opts ...Server
 		}
 	}
 
+	if cfg.session != "" {
+		name = name + "-" + cfg.session
+		cfg.session = name
+	}
+
 	return &Server{
-		nc:    nc,
-		agent: a,
-		cfg:   cfg,
-		owner: owner,
-		name:  name,
+		nc:      nc,
+		handler: h,
+		cfg:     cfg,
+		owner:   owner,
+		name:    name,
 	}, nil
 }
 
@@ -186,6 +198,9 @@ func (s *Server) processPrompt(ctx context.Context, req natsio.Request) {
 		return
 	}
 
+	// Mandatory first message: ack before any latency-inducing work (§6.4).
+	_ = req.Respond(encodeStatusChunk("ack"))
+
 	// Keepalive: emit periodic ack chunks so the caller's inactivity timeout
 	// does not fire during long-running agent work (§6.4).
 	kaCtx, kaCancel := context.WithCancel(ctx)
@@ -203,7 +218,7 @@ func (s *Server) processPrompt(ctx context.Context, req natsio.Request) {
 		}
 	})
 
-	result, runErr := s.agent.Run(ctx, parts...)
+	result, runErr := s.handler.Run(ctx, parts...)
 
 	// Stop keepalive before emitting the response/terminator so ack chunks
 	// never race the terminator or the response chunk.