feat(session): add Fork/Report turn recording for session DAG (TASKS-3 Phase 1)

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 <noreply@anthropic.com>

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` 参照
 
 ## コードの匂い — チェックリスト
 

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)
+	// 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.:
 //

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
+}

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")
+}

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:<taskID>" 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)

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() {

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
+}

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.
@@ -52,6 +53,8 @@ type SubagentManager struct {
 	hasTemperature bool
 	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",

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" に変更