From 18f484165f8ebf13e4fc4294127a59ddd1242e64 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Wed, 4 Mar 2026 22:15:32 +0900 Subject: [PATCH] feat(session): add Fork/Report turn recording for session DAG (TASKS-3 Phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce SessionRecorder interface to bridge pkg/tools → pkg/session without circular imports. SubagentManager now records Fork, Turn, and Completion events in SQLite. processSystemMessage writes TurnReport directly to the store with AdvanceStored to prevent double-writes. Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 8 +- pkg/agent/loop.go | 33 +++- pkg/agent/session_recorder.go | 58 +++++++ pkg/agent/session_recorder_test.go | 245 +++++++++++++++++++++++++++++ pkg/routing/session_key.go | 5 + pkg/session/legacy_adapter.go | 16 ++ pkg/tools/session_recorder.go | 19 +++ pkg/tools/subagent.go | 32 +++- todo/TASKS-3.md | 8 +- 9 files changed, 413 insertions(+), 11 deletions(-) create mode 100644 pkg/agent/session_recorder.go create mode 100644 pkg/agent/session_recorder_test.go create mode 100644 pkg/tools/session_recorder.go diff --git a/CLAUDE.md b/CLAUDE.md index 6576f96b5..31f9e1eaf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,14 +26,16 @@ Lint: `golangci-lint run` - **Sandbox/Spawn**: `pkg/tools/sandbox.go`, `pkg/tools/spawn.go` 実装済み。 - **AgentReporter**: `orch.AgentReporter` / `orch.Noop` / `orch.Broadcaster` で統一。main/heartbeat/subagent 全セッションが同一 Broadcaster に発火。Mini App は `agentLoop.GetOrchBroadcaster()` → `handler.SetOrchBroadcaster()` で受信。 -## Session DAG (Phase 0 実装済み) +## Session DAG (Phase 0–1 実装済み) - **SQLite SessionStore**: `pkg/session/sqlite.go` — `modernc.org/sqlite` (CGO不要)、WAL モード、`sessions` + `turns` テーブル - **SessionStore interface**: `pkg/session/store.go` — Create/Get/List/Append/Turns/Compact/Fork/Prune 等15メソッド -- **LegacyAdapter**: `pkg/session/legacy_adapter.go` — SessionStore をラップし SessionManager と同一 API を提供。`loop.go` 変更なし +- **LegacyAdapter**: `pkg/session/legacy_adapter.go` — SessionStore をラップし SessionManager と同一 API を提供。`Store()` / `AdvanceStored()` で直接 DAG 操作も可能 - **JSON → SQLite migration**: `pkg/session/migrate.go` — 起動時に `sessions/*.json` を検出 → SQLite import → `.json.migrated` にリネーム - **配線**: `pkg/agent/instance.go` の `Sessions` 型が `*LegacyAdapter` に変更。`sessions.db` を workspace 直下に生成 -- **Phase 1以降**: Fork/Report ターン導入、SessionGraph 直接呼び出し、Mini App 可視化 → `todo/TASKS-3.md` 参照 +- **SessionRecorder**: `pkg/tools/session_recorder.go` (interface) + `pkg/agent/session_recorder.go` (impl) — SubagentManager から Fork/Turn/Completion/Report を記録。循環依存回避のためインターフェースは pkg/tools 側 +- **Fork/Report フロー**: `SubagentManager.Spawn()` で Fork 記録、`runTask()` で Turn + Completion 記録、`processSystemMessage()` で TurnReport を直接 store に書き込み + `AdvanceStored` で二重書き込み防止 +- **Phase 2以降**: SessionGraph 直接呼び出し、Compaction、Mini App 可視化 → `todo/TASKS-3.md` 参照 ## コードの匂い — チェックリスト diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 82d9dde26..b474717dd 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -335,6 +335,10 @@ func registerSharedTools( webSearchOpts, ) subagentManager.SetLLMOptions(agent.MaxTokens, agent.Temperature) + // Wire session recorder for DAG persistence. + recorder := newSessionRecorder(agent.Sessions) + conductorKey := routing.BuildAgentMainSessionKey(agent.ID) + subagentManager.SetSessionRecorder(recorder, conductorKey) agent.SubagentMgr = subagentManager spawnTool := tools.NewSpawnTool(subagentManager) currentAgentID := agentID @@ -866,8 +870,25 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe sessionKey := routing.BuildAgentMainSessionKey(agent.ID) historyMsg := fmt.Sprintf("[System: %s] %s", msg.SenderID, msg.Content) - agent.Sessions.AddMessage(sessionKey, "user", historyMsg) - agent.Sessions.MarkDirty(sessionKey) + // Write as TurnReport to the store for DAG tracking, with legacy fallback. + subagentSessionKey := routing.BuildSubagentSessionKey(extractTaskID(msg.SenderID)) + store := agent.Sessions.Store() + reportTurn := &session.Turn{ + Kind: session.TurnReport, + OriginKey: subagentSessionKey, + Author: msg.SenderID, + Messages: []providers.Message{{Role: "user", Content: historyMsg}}, + } + if err := store.Append(sessionKey, reportTurn); err != nil { + logger.ErrorCF("agent", "Failed to record report turn, falling back to legacy", + map[string]any{"error": err.Error()}) + agent.Sessions.AddMessage(sessionKey, "user", historyMsg) + agent.Sessions.MarkDirty(sessionKey) + } else { + // Update in-memory cache so conductor sees the message on next turn. + agent.Sessions.AddFullMessage(sessionKey, providers.Message{Role: "user", Content: historyMsg}) + agent.Sessions.AdvanceStored(sessionKey, 1) + } // Send a brief notification (SkipPlaceholder to avoid corrupting status messages) label := msg.SenderID @@ -897,6 +918,14 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe return "", nil } +// extractTaskID extracts the task ID from a sender ID like "subagent:subagent-1". +func extractTaskID(senderID string) string { + if idx := strings.LastIndex(senderID, ":"); idx >= 0 { + return senderID[idx+1:] + } + return senderID +} + // formatSubagentCompletion builds the user-facing notification for a completed subagent. // If metadata contains duration_ms and tool_calls it produces e.g.: // diff --git a/pkg/agent/session_recorder.go b/pkg/agent/session_recorder.go new file mode 100644 index 000000000..b654911e3 --- /dev/null +++ b/pkg/agent/session_recorder.go @@ -0,0 +1,58 @@ +package agent + +import ( + "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/session" + "github.com/sipeed/picoclaw/pkg/tools" +) + +// sessionRecorderImpl bridges tools.SessionRecorder → session.SessionStore. +type sessionRecorderImpl struct { + adapter *session.LegacyAdapter +} + +var _ tools.SessionRecorder = (*sessionRecorderImpl)(nil) + +func newSessionRecorder(adapter *session.LegacyAdapter) *sessionRecorderImpl { + return &sessionRecorderImpl{adapter: adapter} +} + +func (r *sessionRecorderImpl) RecordFork(conductorKey, subagentKey, taskID, label string) error { + store := r.adapter.Store() + return store.Fork(conductorKey, subagentKey, &session.CreateOpts{ + ForkTurnID: taskID, + Label: label, + }) +} + +func (r *sessionRecorderImpl) RecordSubagentTurn(subagentKey string, messages []providers.Message) error { + store := r.adapter.Store() + turn := &session.Turn{ + Kind: session.TurnNormal, + Messages: messages, + } + return store.Append(subagentKey, turn) +} + +func (r *sessionRecorderImpl) RecordCompletion(subagentKey, status, result string) error { + store := r.adapter.Store() + return store.SetStatus(subagentKey, status) +} + +func (r *sessionRecorderImpl) RecordReport(conductorKey, subagentKey, senderID, content string) error { + store := r.adapter.Store() + turn := &session.Turn{ + Kind: session.TurnReport, + OriginKey: subagentKey, + Author: senderID, + Messages: []providers.Message{ + {Role: "user", Content: content}, + }, + } + if err := store.Append(conductorKey, turn); err != nil { + return err + } + // Advance LegacyAdapter's stored counter so flush loop doesn't double-write. + r.adapter.AdvanceStored(conductorKey, 1) + return nil +} diff --git a/pkg/agent/session_recorder_test.go b/pkg/agent/session_recorder_test.go new file mode 100644 index 000000000..410eb59dc --- /dev/null +++ b/pkg/agent/session_recorder_test.go @@ -0,0 +1,245 @@ +package agent + +import ( + "os" + "path/filepath" + "testing" + + "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/session" +) + +func newTestRecorder(t *testing.T) (*sessionRecorderImpl, *session.LegacyAdapter, session.SessionStore) { + t.Helper() + tmpDir := t.TempDir() + dbPath := filepath.Join(tmpDir, "test.db") + store, err := session.OpenSQLiteStore(dbPath) + if err != nil { + t.Fatalf("open store: %v", err) + } + adapter := session.NewLegacyAdapter(store) + t.Cleanup(func() { adapter.Close() }) + recorder := newSessionRecorder(adapter) + return recorder, adapter, store +} + +func TestRecordFork(t *testing.T) { + rec, _, store := newTestRecorder(t) + + // Create conductor session first. + if err := store.Create("conductor:main", nil); err != nil { + t.Fatalf("create conductor session: %v", err) + } + + err := rec.RecordFork("conductor:main", "subagent:subagent-1", "subagent-1", "scout") + if err != nil { + t.Fatalf("RecordFork: %v", err) + } + + // Verify child session exists with correct parent. + info, err := store.Get("subagent:subagent-1") + if err != nil { + t.Fatalf("Get child: %v", err) + } + if info == nil { + t.Fatal("child session not found") + } + if info.ParentKey != "conductor:main" { + t.Errorf("ParentKey = %q, want %q", info.ParentKey, "conductor:main") + } + if info.ForkTurnID != "subagent-1" { + t.Errorf("ForkTurnID = %q, want %q", info.ForkTurnID, "subagent-1") + } + if info.Label != "scout" { + t.Errorf("Label = %q, want %q", info.Label, "scout") + } + + // Verify parent lists child. + children, err := store.Children("conductor:main") + if err != nil { + t.Fatalf("Children: %v", err) + } + if len(children) != 1 { + t.Fatalf("children count = %d, want 1", len(children)) + } + if children[0].Key != "subagent:subagent-1" { + t.Errorf("child key = %q, want %q", children[0].Key, "subagent:subagent-1") + } +} + +func TestRecordSubagentTurn(t *testing.T) { + rec, _, store := newTestRecorder(t) + + // Create subagent session. + if err := store.Create("subagent:subagent-1", nil); err != nil { + t.Fatalf("create: %v", err) + } + + msgs := []providers.Message{ + {Role: "system", Content: "You are a scout."}, + {Role: "user", Content: "Investigate X."}, + {Role: "assistant", Content: "Found Y."}, + } + if err := rec.RecordSubagentTurn("subagent:subagent-1", msgs); err != nil { + t.Fatalf("RecordSubagentTurn: %v", err) + } + + turns, err := store.Turns("subagent:subagent-1", 0) + if err != nil { + t.Fatalf("Turns: %v", err) + } + if len(turns) != 1 { + t.Fatalf("turns count = %d, want 1", len(turns)) + } + if turns[0].Kind != session.TurnNormal { + t.Errorf("Kind = %d, want TurnNormal", turns[0].Kind) + } + if len(turns[0].Messages) != 3 { + t.Errorf("messages count = %d, want 3", len(turns[0].Messages)) + } + if turns[0].Messages[2].Content != "Found Y." { + t.Errorf("last message = %q, want %q", turns[0].Messages[2].Content, "Found Y.") + } +} + +func TestRecordCompletion(t *testing.T) { + rec, _, store := newTestRecorder(t) + + if err := store.Create("subagent:subagent-1", nil); err != nil { + t.Fatalf("create: %v", err) + } + + if err := rec.RecordCompletion("subagent:subagent-1", "completed", "done"); err != nil { + t.Fatalf("RecordCompletion: %v", err) + } + + info, err := store.Get("subagent:subagent-1") + if err != nil { + t.Fatalf("Get: %v", err) + } + if info.Status != "completed" { + t.Errorf("Status = %q, want %q", info.Status, "completed") + } + + // Test failed status. + if err := store.Create("subagent:subagent-2", nil); err != nil { + t.Fatalf("create: %v", err) + } + if err := rec.RecordCompletion("subagent:subagent-2", "failed", "error"); err != nil { + t.Fatalf("RecordCompletion failed: %v", err) + } + info2, _ := store.Get("subagent:subagent-2") + if info2.Status != "failed" { + t.Errorf("Status = %q, want %q", info2.Status, "failed") + } +} + +func TestRecordReport(t *testing.T) { + rec, adapter, store := newTestRecorder(t) + + // Create conductor session via adapter so it's in cache. + _ = adapter.GetOrCreate("conductor:main") + + if err := store.Create("subagent:subagent-1", nil); err != nil { + t.Fatalf("create subagent: %v", err) + } + + content := "[System: subagent:subagent-1] Task 'scout' completed.\n\nResult:\nFound Y." + if err := rec.RecordReport("conductor:main", "subagent:subagent-1", "subagent:subagent-1", content); err != nil { + t.Fatalf("RecordReport: %v", err) + } + + // Verify TurnReport in store. + turns, err := store.Turns("conductor:main", 0) + if err != nil { + t.Fatalf("Turns: %v", err) + } + if len(turns) != 1 { + t.Fatalf("turns count = %d, want 1", len(turns)) + } + if turns[0].Kind != session.TurnReport { + t.Errorf("Kind = %d, want TurnReport(%d)", turns[0].Kind, session.TurnReport) + } + if turns[0].OriginKey != "subagent:subagent-1" { + t.Errorf("OriginKey = %q, want %q", turns[0].OriginKey, "subagent:subagent-1") + } + if turns[0].Author != "subagent:subagent-1" { + t.Errorf("Author = %q, want %q", turns[0].Author, "subagent:subagent-1") + } + if len(turns[0].Messages) != 1 || turns[0].Messages[0].Role != "user" { + t.Errorf("unexpected messages: %v", turns[0].Messages) + } +} + +func TestAdvanceStoredPreventsDoubleWrite(t *testing.T) { + rec, adapter, store := newTestRecorder(t) + + // Create conductor session via adapter. + _ = adapter.GetOrCreate("conductor:main") + + if err := store.Create("subagent:subagent-1", nil); err != nil { + t.Fatalf("create subagent: %v", err) + } + + // Simulate: conductor has 2 messages already flushed. + adapter.AddMessage("conductor:main", "user", "hello") + adapter.AddMessage("conductor:main", "assistant", "hi") + if err := adapter.Save("conductor:main"); err != nil { + t.Fatalf("Save: %v", err) + } + + // RecordReport writes directly to store and advances stored counter. + content := "[System: subagent:subagent-1] result" + if err := rec.RecordReport("conductor:main", "subagent:subagent-1", "subagent:subagent-1", content); err != nil { + t.Fatalf("RecordReport: %v", err) + } + + // The in-memory cache should also be updated (by loop.go calling AddFullMessage). + // Simulate what loop.go does after RecordReport succeeds. + adapter.AddFullMessage("conductor:main", providers.Message{Role: "user", Content: content}) + // AdvanceStored was already called by RecordReport, so stored = 3 + 1 = 4 + // but we added 1 message to cache making it len=4 as well. No double write. + + // Save should NOT re-write the report turn. + if err := adapter.Save("conductor:main"); err != nil { + t.Fatalf("Save after report: %v", err) + } + + // Count all turns in store for conductor session. + turns, err := store.Turns("conductor:main", 0) + if err != nil { + t.Fatalf("Turns: %v", err) + } + + // Expected: turn 1 (initial 2 msgs), turn 2 (TurnReport from RecordReport) + // NOT turn 3 (duplicate from flush). + if len(turns) != 2 { + t.Errorf("turns count = %d, want 2 (no double-write)", len(turns)) + for i, turn := range turns { + t.Logf(" turn[%d]: seq=%d kind=%d msgs=%d", i, turn.Seq, turn.Kind, len(turn.Messages)) + } + } +} + +func TestExtractTaskID(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"subagent:subagent-1", "subagent-1"}, + {"subagent:subagent-42", "subagent-42"}, + {"plain-id", "plain-id"}, + {"a:b:c", "c"}, + } + for _, tt := range tests { + got := extractTaskID(tt.input) + if got != tt.want { + t.Errorf("extractTaskID(%q) = %q, want %q", tt.input, got, tt.want) + } + } +} + +func init() { + // Suppress log output in tests. + os.Setenv("PICOCLAW_LOG_LEVEL", "error") +} diff --git a/pkg/routing/session_key.go b/pkg/routing/session_key.go index eab592bec..19e6dd390 100644 --- a/pkg/routing/session_key.go +++ b/pkg/routing/session_key.go @@ -42,6 +42,11 @@ func BuildAgentMainSessionKey(agentID string) string { return fmt.Sprintf("agent:%s:%s", NormalizeAgentID(agentID), DefaultMainKey) } +// BuildSubagentSessionKey returns "subagent:" for subagent sessions. +func BuildSubagentSessionKey(taskID string) string { + return fmt.Sprintf("subagent:%s", taskID) +} + // BuildAgentPeerSessionKey constructs a session key based on agent, channel, peer, and DM scope. func BuildAgentPeerSessionKey(params SessionKeyParams) string { agentID := NormalizeAgentID(params.AgentID) diff --git a/pkg/session/legacy_adapter.go b/pkg/session/legacy_adapter.go index 9440a5915..67b05cd1f 100644 --- a/pkg/session/legacy_adapter.go +++ b/pkg/session/legacy_adapter.go @@ -459,6 +459,22 @@ func (la *LegacyAdapter) Save(key string) error { return nil } +// Store returns the underlying SessionStore for direct DAG operations. +func (la *LegacyAdapter) Store() SessionStore { + return la.store +} + +// AdvanceStored increments the stored counter for a session by delta, +// preventing the flush loop from re-persisting messages already written +// directly to the store (e.g. TurnReport). +func (la *LegacyAdapter) AdvanceStored(key string, delta int) { + la.mu.Lock() + defer la.mu.Unlock() + if c, ok := la.cache[key]; ok { + c.stored += delta + } +} + // Close stops the background flush loop and persists all dirty sessions. func (la *LegacyAdapter) Close() { diff --git a/pkg/tools/session_recorder.go b/pkg/tools/session_recorder.go new file mode 100644 index 000000000..1ac339ecb --- /dev/null +++ b/pkg/tools/session_recorder.go @@ -0,0 +1,19 @@ +package tools + +import "github.com/sipeed/picoclaw/pkg/providers" + +// SessionRecorder records session DAG events (fork, turns, completion, report). +// Implemented by pkg/agent to bridge tools → session without circular imports. +type SessionRecorder interface { + // RecordFork creates a child session forked from the conductor session. + RecordFork(conductorSessionKey, subagentSessionKey, taskID, label string) error + + // RecordSubagentTurn records the initial + final messages of a subagent run. + RecordSubagentTurn(subagentSessionKey string, messages []providers.Message) error + + // RecordCompletion marks a subagent session as completed/failed. + RecordCompletion(subagentSessionKey, status, result string) error + + // RecordReport injects a TurnReport into the conductor session. + RecordReport(conductorSessionKey, subagentSessionKey, senderID, content string) error +} diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index e3b7435bf..b5266345f 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -12,6 +12,7 @@ import ( "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/orch" "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/routing" ) // spawnTimeout is the hard upper bound for a single spawn goroutine. @@ -50,8 +51,10 @@ type SubagentManager struct { temperature float64 hasMaxTokens bool hasTemperature bool - nextID int - reporter orch.AgentReporter + nextID int + reporter orch.AgentReporter + recorder SessionRecorder + conductorSessionKey string } func NewSubagentManager( @@ -96,6 +99,14 @@ func (sm *SubagentManager) SetTools(tools *ToolRegistry) { sm.tools = tools } +// SetSessionRecorder configures session recording for DAG persistence. +func (sm *SubagentManager) SetSessionRecorder(r SessionRecorder, conductorSessionKey string) { + sm.mu.Lock() + defer sm.mu.Unlock() + sm.recorder = r + sm.conductorSessionKey = conductorSessionKey +} + // RegisterTool registers a tool for subagent execution. func (sm *SubagentManager) RegisterTool(tool Tool) { sm.mu.Lock() @@ -128,6 +139,12 @@ func (sm *SubagentManager) Spawn( sm.reporter.ReportSpawn(taskID, label, task) + // Record fork in session DAG (conductor → subagent). + if sm.recorder != nil && sm.conductorSessionKey != "" { + subSessionKey := routing.BuildSubagentSessionKey(taskID) + _ = sm.recorder.RecordFork(sm.conductorSessionKey, subSessionKey, taskID, label) + } + // Start task in background with a detached context that has a hard timeout. // The spawned goroutine must outlive the parent (e.g. heartbeat session) // which may finish before the subagent completes. @@ -252,6 +269,11 @@ After completing, provide a clear summary of what was done and how it was verifi gcReason = "canceled" } sm.reporter.ReportGC(task.ID, gcReason) + // Record failure/cancellation in session DAG. + if sm.recorder != nil { + subKey := routing.BuildSubagentSessionKey(task.ID) + _ = sm.recorder.RecordCompletion(subKey, task.Status, task.Result) + } result = &ToolResult{ ForLLM: task.Result, ForUser: "", @@ -270,6 +292,12 @@ After completing, provide a clear summary of what was done and how it was verifi // Notify conductor of the result sm.reporter.ReportConversation(task.ID, "conductor", loopResult.Content) sm.reporter.ReportGC(task.ID, "completed") + // Record subagent turn and completion in session DAG. + if sm.recorder != nil { + subKey := routing.BuildSubagentSessionKey(task.ID) + _ = sm.recorder.RecordSubagentTurn(subKey, messages) + _ = sm.recorder.RecordCompletion(subKey, "completed", loopResult.Content) + } result = &ToolResult{ ForLLM: fmt.Sprintf( "Subagent '%s' completed (iterations: %d, tool calls: %d): %s", diff --git a/todo/TASKS-3.md b/todo/TASKS-3.md index c87060af9..6d5225ebe 100644 --- a/todo/TASKS-3.md +++ b/todo/TASKS-3.md @@ -171,18 +171,18 @@ func (tw *TurnWriter) Discard() --- -### Phase 1: Fork/Report ターン導入 +### Phase 1: Fork/Report ターン導入 ✅ -5. **Fork 操作** +5. ✅ **Fork 操作** - `SessionStore.Fork()` 実装 - 親セッションに `TurnForkPoint` を追記、子セッションを `parent_key` 付きで作成 -6. **Report ターン** +6. ✅ **Report ターン** - `TurnReport` の Append/Turns/Messages 対応 - `origin_key` でどのセッションの報告かを追跡 - user role でメッセージ格納 (system role 禁止) -7. **サブエージェントセッション永続化** +7. ✅ **サブエージェントセッション永続化** - サブエージェント実行中のターンを SQLite に記録 - 完了後にセッション status を "completed" に変更