opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(subturn): align types with pkg/session & pkg/tools, use SessionStore for ephemeral

Browse source 
- Remove duplicated type definitions (ToolResult, Session, Message) → use official types:
  - tools.ToolResult (from pkg/tools/result.go)
  - session.SessionStore + providers.Message (from pkg/session/session_store.go)
- Change turnState.session from *Session to session.SessionStore
  - Ephemeral session now uses a pure in-memory implementation (no persistence)
- Prepare runTurn() for real integration:
  - Placeholder remains, but designed to accept AgentInstance
  - Ready to hook into runAgentLoop / runLLMIteration once steering/turnState lands
- Rebased onto refactor/agent (steering merged), resolves type mismatches

This makes the sub-turn PoC more compatible with the ongoing agent refactor (#1316).
This commit is contained in:
Administrator 2026-03-16 13:57:46 +08:00
parent 296b5128df
commit 106f1a5bda
2 changed files with 151 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

162
pkg/agent/subturn.go
View file

@ -7,6 +7,8 @@ import (
"sync"
"sync/atomic"
"github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/session"
"github.com/sipeed/picoclaw/pkg/tools"
)
@ -36,20 +38,20 @@ type SubTurnSpawnEvent struct {
type SubTurnEndEvent struct {
ChildID string
Result *ToolResult
Result *tools.ToolResult
Err error
}
type SubTurnResultDeliveredEvent struct {
ParentID string
ChildID string
Result *ToolResult
Result *tools.ToolResult
}
type SubTurnOrphanResultEvent struct {
ParentID string
ChildID string
Result *ToolResult
Result *tools.ToolResult
}
// ====================== turnState (Simplified, reusable with existing structs) ======================
@ -60,8 +62,8 @@ type turnState struct {
parentTurnID string
depth int
childTurnIDs []string
pendingResults chan *ToolResult
session *Session
pendingResults chan *tools.ToolResult
session session.SessionStore
mu sync.Mutex
isFinished bool // Marks if the parent Turn has ended
}
@ -86,7 +88,7 @@ func newTurnState(ctx context.Context, id string, parent *turnState) *turnState
// Under high concurrency or long-running sub-turns, this might fill up and cause
// intermediate results to be discarded in deliverSubTurnResult.
// For production, consider an unbounded queue or a blocking strategy with backpressure.
pendingResults: make(chan *ToolResult, 16),
pendingResults: make(chan *tools.ToolResult, 16),
}
}
@ -100,16 +102,70 @@ func (ts *turnState) Finish() {
}
}
// newEphemeralSession - Pure in-memory temporary Session (avoids polluting the main session)
func newEphemeralSession(parent *Session) *Session {
// In a real project, it's recommended to copy only necessary fields; simplified here.
return &Session{
History: make([]Message, 0, len(parent.History)),
// ephemeralSessionStore is a pure in-memory SessionStore for SubTurns.
// It never writes to disk, keeping sub-turn history isolated from the parent session.
type ephemeralSessionStore struct {
mu sync.Mutex
history []providers.Message
summary string
}
func (e *ephemeralSessionStore) AddMessage(sessionKey, role, content string) {
e.mu.Lock()
defer e.mu.Unlock()
e.history = append(e.history, providers.Message{Role: role, Content: content})
}
func (e *ephemeralSessionStore) AddFullMessage(sessionKey string, msg providers.Message) {
e.mu.Lock()
defer e.mu.Unlock()
e.history = append(e.history, msg)
}
func (e *ephemeralSessionStore) GetHistory(key string) []providers.Message {
e.mu.Lock()
defer e.mu.Unlock()
out := make([]providers.Message, len(e.history))
copy(out, e.history)
return out
}
func (e *ephemeralSessionStore) GetSummary(key string) string {
e.mu.Lock()
defer e.mu.Unlock()
return e.summary
}
func (e *ephemeralSessionStore) SetSummary(key, summary string) {
e.mu.Lock()
defer e.mu.Unlock()
e.summary = summary
}
func (e *ephemeralSessionStore) SetHistory(key string, history []providers.Message) {
e.mu.Lock()
defer e.mu.Unlock()
e.history = make([]providers.Message, len(history))
copy(e.history, history)
}
func (e *ephemeralSessionStore) TruncateHistory(key string, keepLast int) {
e.mu.Lock()
defer e.mu.Unlock()
if len(e.history) > keepLast {
e.history = e.history[len(e.history)-keepLast:]
}
}
func (e *ephemeralSessionStore) Save(key string) error { return nil }
func (e *ephemeralSessionStore) Close() error { return nil }
func newEphemeralSession(_ session.SessionStore) session.SessionStore {
return &ephemeralSessionStore{}
}
// ====================== Core Function: spawnSubTurn ======================
func spawnSubTurn(ctx context.Context, parentTS *turnState, cfg SubTurnConfig) (result *ToolResult, err error) {
func spawnSubTurn(ctx context.Context, al *AgentLoop, parentTS *turnState, cfg SubTurnConfig) (result *tools.ToolResult, err error) {
// 1. Depth limit check
if parentTS.depth >= maxSubTurnDepth {
return nil, ErrDepthLimitExceeded
@ -153,9 +209,9 @@ func spawnSubTurn(ctx context.Context, parentTS *turnState, cfg SubTurnConfig) (
})
}()
// 7. Execute full runTurn (follows the main execution path, all hooks, steering, and interrupts are effective!)
// Pass the childCtx so the sub-turn can be interrupted if the parent is cancelled.
result, err = runTurn(childCtx, childTS, childTS.session, cfg)
// 7. Execute sub-turn via the real agent loop.
// Build a child AgentInstance from SubTurnConfig, inheriting defaults from the parent agent.
result, err = runTurn(childCtx, al, childTS, cfg)
// 8. Deliver result back to parent Turn
deliverSubTurnResult(parentTS, childID, result)
@ -164,7 +220,7 @@ func spawnSubTurn(ctx context.Context, parentTS *turnState, cfg SubTurnConfig) (
}
// ====================== Result Delivery ======================
func deliverSubTurnResult(parentTS *turnState, childID string, result *ToolResult) {
func deliverSubTurnResult(parentTS *turnState, childID string, result *tools.ToolResult) {
parentTS.mu.Lock()
defer parentTS.mu.Unlock()
@ -196,26 +252,58 @@ func deliverSubTurnResult(parentTS *turnState, childID string, result *ToolResul
}
}
// ====================== Placeholder Function (Actually reuses runTurn in loop.go) ======================
func runTurn(ctx context.Context, ts *turnState, session *Session, cfg SubTurnConfig) (*ToolResult, error) {
// TODO: Directly call the existing runTurn implementation in your project here
// Ensure the existing runTurn respects the context for cancellation.
return &ToolResult{Content: "Sub-turn executed successfully"}, nil
// runTurn builds a temporary AgentInstance from SubTurnConfig and delegates to
// the real agent loop. The child's ephemeral session is used for history so it
// never pollutes the parent session.
func runTurn(ctx context.Context, al *AgentLoop, ts *turnState, cfg SubTurnConfig) (*tools.ToolResult, error) {
// Derive candidates from the requested model using the parent loop's provider.
defaultProvider := al.GetConfig().Agents.Defaults.Provider
candidates := providers.ResolveCandidates(
providers.ModelConfig{Primary: cfg.Model},
defaultProvider,
)
// Build a minimal AgentInstance for this sub-turn.
// It reuses the parent loop's provider and config, but gets its own
// ephemeral session store and tool registry.
toolRegistry := tools.NewToolRegistry()
for _, t := range cfg.Tools {
toolRegistry.Register(t)
}
parentAgent := al.GetRegistry().GetDefaultAgent()
childAgent := &AgentInstance{
ID: ts.turnID,
Model: cfg.Model,
MaxIterations: parentAgent.MaxIterations,
MaxTokens: cfg.MaxTokens,
Temperature: parentAgent.Temperature,
ThinkingLevel: parentAgent.ThinkingLevel,
ContextWindow: cfg.MaxTokens,
SummarizeMessageThreshold: parentAgent.SummarizeMessageThreshold,
SummarizeTokenPercent: parentAgent.SummarizeTokenPercent,
Provider: parentAgent.Provider,
Sessions: ts.session,
ContextBuilder: parentAgent.ContextBuilder,
Tools: toolRegistry,
Candidates: candidates,
}
if childAgent.MaxTokens == 0 {
childAgent.MaxTokens = parentAgent.MaxTokens
childAgent.ContextWindow = parentAgent.ContextWindow
}
finalContent, err := al.runAgentLoop(ctx, childAgent, processOptions{
SessionKey: ts.turnID,
UserMessage: cfg.SystemPrompt,
DefaultResponse: "",
EnableSummary: false,
SendResponse: false,
})
if err != nil {
return nil, err
}
return &tools.ToolResult{ForLLM: finalContent}, nil
}
// ====================== Other Types (Reused or simplified from existing code) ======================
type ToolResult struct {
Content string
}
func (r *ToolResult) ToMessage() Message {
return Message{Content: r.Content}
}
type Session struct {
History []Message
}
type Message struct {
Content string
}
// ====================== Other Types ======================

41
pkg/agent/subturn_test.go
View file

@ -96,6 +96,9 @@ func TestSpawnSubTurn(t *testing.T) {
},
}
al, _, _, _, cleanup := newTestAgentLoop(t)
defer cleanup()
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Prepare parent Turn
@ -104,8 +107,8 @@ func TestSpawnSubTurn(t *testing.T) {
turnID: "parent-1",
depth: tt.parentDepth,
childTurnIDs: []string{},
pendingResults: make(chan *ToolResult, 10),
session: &Session{History: []Message{}},
pendingResults: make(chan *tools.ToolResult, 10),
session: &ephemeralSessionStore{},
}
// Replace mock with test collector
@ -115,7 +118,7 @@ func TestSpawnSubTurn(t *testing.T) {
defer func() { MockEventBus.Emit = originalEmit }()
// Execute spawnSubTurn
result, err := spawnSubTurn(context.Background(), parent, tt.config)
result, err := spawnSubTurn(context.Background(), al, parent, tt.config)
// Assert errors
if tt.wantErr != nil {
@ -152,7 +155,7 @@ func TestSpawnSubTurn(t *testing.T) {
}
// Verify result delivery (pendingResults or history)
if len(parent.pendingResults) > 0 || len(parent.session.History) > 0 {
if len(parent.pendingResults) > 0 || len(parent.session.GetHistory("")) > 0 {
// Result delivered via at least one path
} else {
t.Error("child result not delivered")
@ -163,39 +166,47 @@ func TestSpawnSubTurn(t *testing.T) {
// ====================== Extra Independent Test: Ephemeral Session Isolation ======================
func TestSpawnSubTurn_EphemeralSessionIsolation(t *testing.T) {
al, _, _, _, cleanup := newTestAgentLoop(t)
defer cleanup()
parentSession := &ephemeralSessionStore{}
parentSession.AddMessage("", "user", "parent msg")
parent := &turnState{
ctx: context.Background(),
turnID: "parent-1",
depth: 0,
session: &Session{History: []Message{{Content: "parent msg"}}},
session: parentSession,
}
cfg := SubTurnConfig{Model: "gpt-4o-mini", Tools: []tools.Tool{}}
// Record main session length before execution
originalLen := len(parent.session.History)
originalLen := len(parent.session.GetHistory(""))
_, _ = spawnSubTurn(context.Background(), parent, cfg)
_, _ = spawnSubTurn(context.Background(), al, parent, cfg)
// After sub-turn ends, main session must remain unchanged
if len(parent.session.History) != originalLen {
if len(parent.session.GetHistory("")) != originalLen {
t.Error("ephemeral session polluted the main session")
}
}
// ====================== Extra Independent Test: Result Delivery Path ======================
func TestSpawnSubTurn_ResultDelivery(t *testing.T) {
al, _, _, _, cleanup := newTestAgentLoop(t)
defer cleanup()
parent := &turnState{
ctx: context.Background(),
turnID: "parent-1",
depth: 0,
pendingResults: make(chan *ToolResult, 1),
session: &Session{},
pendingResults: make(chan *tools.ToolResult, 1),
session: &ephemeralSessionStore{},
}
cfg := SubTurnConfig{Model: "gpt-4o-mini", Tools: []tools.Tool{}}
_, _ = spawnSubTurn(context.Background(), parent, cfg)
_, _ = spawnSubTurn(context.Background(), al, parent, cfg)
// Check if pendingResults received the result
select {
@ -216,8 +227,8 @@ func TestSpawnSubTurn_OrphanResultRouting(t *testing.T) {
cancelFunc: cancelParent,
turnID: "parent-1",
depth: 0,
pendingResults: make(chan *ToolResult, 1),
session: &Session{History: []Message{}},
pendingResults: make(chan *tools.ToolResult, 1),
session: &ephemeralSessionStore{},
}
collector := &eventCollector{}
@ -229,7 +240,7 @@ func TestSpawnSubTurn_OrphanResultRouting(t *testing.T) {
parent.Finish()
// Call deliverSubTurnResult directly to simulate a delayed child
deliverSubTurnResult(parent, "delayed-child", &ToolResult{Content: "late result"})
deliverSubTurnResult(parent, "delayed-child", &tools.ToolResult{ForLLM: "late result"})
// Verify Orphan event is emitted
if !collector.hasEventOfType(SubTurnOrphanResultEvent{}) {
@ -237,7 +248,7 @@ func TestSpawnSubTurn_OrphanResultRouting(t *testing.T) {
}
// Verify history is NOT polluted
if len(parent.session.History) != 0 {
if len(parent.session.GetHistory("")) != 0 {
t.Error("Parent history was polluted by orphan result")
}
}