merge: integrate main into refactor-inbound-context-routing-session

2026-06-12 18:08:54 +00:00 · 2026-04-13 13:25:07 +08:00
parent 0f23535165 2b2bc26f8e
commit 0c6ad33a9c
134 changed files with 13291 additions and 1321 deletions
@@ -1398,6 +1398,40 @@ func (m *toolFeedbackProvider) GetDefaultModel() string {
 	return "heartbeat-tool-feedback-model"
 }

+type picoInterleavedContentProvider struct {
+	calls int
+}
+
+func (m *picoInterleavedContentProvider) Chat(
+	ctx context.Context,
+	messages []providers.Message,
+	tools []providers.ToolDefinition,
+	model string,
+	opts map[string]any,
+) (*providers.LLMResponse, error) {
+	m.calls++
+	if m.calls == 1 {
+		return &providers.LLMResponse{
+			Content: "intermediate model text",
+			ToolCalls: []providers.ToolCall{{
+				ID:        "call_tool_limit_test",
+				Type:      "function",
+				Name:      "tool_limit_test_tool",
+				Arguments: map[string]any{"value": "x"},
+			}},
+		}, nil
+	}
+
+	return &providers.LLMResponse{
+		Content:   "final model text",
+		ToolCalls: []providers.ToolCall{},
+	}, nil
+}
+
+func (m *picoInterleavedContentProvider) GetDefaultModel() string {
+	return "pico-interleaved-content-model"
+}
+
 type toolLimitOnlyProvider struct{}

 func (m *toolLimitOnlyProvider) Chat(
@@ -2229,7 +2263,7 @@ func TestProcessMessage_FallbackUsesPerCandidateProvider(t *testing.T) {
 			},
 			{
 				ModelName: "gemma-fallback",
-				Model:     "gemini/gemma-3-27b-it",
+				Model:     "openrouter/gemma-3-27b-it",
 				APIBase:   fallbackServer.URL,
 				APIKeys:   config.SimpleSecureStrings("fallback-key"),
 				Workspace: workspace,
@@ -2970,6 +3004,66 @@ func TestProcessMessage_PublishesReasoningContentToReasoningChannel(t *testing.T
 	}
 }

+func TestProcessMessage_PicoPublishesReasoningAsThoughtMessage(t *testing.T) {
+	tmpDir := t.TempDir()
+	cfg := &config.Config{
+		Agents: config.AgentsConfig{
+			Defaults: config.AgentDefaults{
+				Workspace:         tmpDir,
+				ModelName:         "test-model",
+				MaxTokens:         4096,
+				MaxToolIterations: 10,
+			},
+		},
+	}
+
+	msgBus := bus.NewMessageBus()
+	provider := &reasoningContentProvider{
+		response:         "final answer",
+		reasoningContent: "thinking trace",
+	}
+	al := NewAgentLoop(cfg, msgBus, provider)
+
+	response, err := al.processMessage(context.Background(), bus.InboundMessage{
+		Channel:  "pico",
+		SenderID: "user1",
+		ChatID:   "pico:test-session",
+		Content:  "hello",
+	})
+	if err != nil {
+		t.Fatalf("processMessage() error = %v", err)
+	}
+	if response != "final answer" {
+		t.Fatalf("processMessage() response = %q, want %q", response, "final answer")
+	}
+
+	var thoughtMsg *bus.OutboundMessage
+	deadline := time.After(3 * time.Second)
+
+	for thoughtMsg == nil {
+		select {
+		case outbound := <-msgBus.OutboundChan():
+			msg := outbound
+			if msg.Content == "thinking trace" {
+				thoughtMsg = &msg
+			}
+		case <-deadline:
+			t.Fatal("expected thought outbound message for pico")
+		}
+	}
+
+	if thoughtMsg.Channel != "pico" || thoughtMsg.ChatID != "pico:test-session" {
+		t.Fatalf("thought message route = %s/%s, want pico/pico:test-session", thoughtMsg.Channel, thoughtMsg.ChatID)
+	}
+	if thoughtMsg.Context.Raw[metadataKeyMessageKind] != messageKindThought {
+		t.Fatalf(
+			"thought metadata kind = %q, want %q",
+			thoughtMsg.Context.Raw[metadataKeyMessageKind],
+			messageKindThought,
+		)
+	}
+}
+
 func TestProcessHeartbeat_DoesNotPublishToolFeedback(t *testing.T) {
 	tmpDir := t.TempDir()
 	heartbeatFile := filepath.Join(tmpDir, "heartbeat-task.txt")
@@ -3135,6 +3229,133 @@ func TestProcessMessage_MessageToolPublishesOutboundWithTurnMetadata(t *testing.
 	}
 }

+func TestRun_PicoPublishesAssistantContentDuringToolCallsWithoutFinalDuplicate(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	cfg := &config.Config{
+		Agents: config.AgentsConfig{
+			Defaults: config.AgentDefaults{
+				Workspace:         tmpDir,
+				ModelName:         "test-model",
+				MaxTokens:         4096,
+				MaxToolIterations: 10,
+			},
+		},
+	}
+
+	msgBus := bus.NewMessageBus()
+	provider := &picoInterleavedContentProvider{}
+	al := NewAgentLoop(cfg, msgBus, provider)
+
+	agent := al.GetRegistry().GetDefaultAgent()
+	if agent == nil {
+		t.Fatal("expected default agent")
+	}
+	agent.Tools.Register(&toolLimitTestTool{})
+
+	runCtx, runCancel := context.WithCancel(context.Background())
+	defer runCancel()
+
+	runDone := make(chan error, 1)
+	go func() {
+		runDone <- al.Run(runCtx)
+	}()
+
+	if err := msgBus.PublishInbound(context.Background(), bus.InboundMessage{
+		Channel:  "pico",
+		SenderID: "user-1",
+		ChatID:   "session-1",
+		Content:  "run with tools",
+	}); err != nil {
+		t.Fatalf("PublishInbound() error = %v", err)
+	}
+
+	outputs := make([]string, 0, 2)
+	deadline := time.After(2 * time.Second)
+	for len(outputs) < 2 {
+		select {
+		case outbound := <-msgBus.OutboundChan():
+			outputs = append(outputs, outbound.Content)
+		case <-deadline:
+			t.Fatalf("timed out waiting for pico outputs, got %v", outputs)
+		}
+	}
+
+	if outputs[0] != "intermediate model text" {
+		t.Fatalf("first outbound content = %q, want %q", outputs[0], "intermediate model text")
+	}
+	if outputs[1] != "final model text" {
+		t.Fatalf("second outbound content = %q, want %q", outputs[1], "final model text")
+	}
+
+	runCancel()
+	select {
+	case err := <-runDone:
+		if err != nil {
+			t.Fatalf("Run() error = %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for Run() to exit")
+	}
+
+	select {
+	case outbound := <-msgBus.OutboundChan():
+		if outbound.Content == "final model text" {
+			t.Fatalf("unexpected duplicate final pico output: %+v", outbound)
+		}
+	case <-time.After(200 * time.Millisecond):
+	}
+}
+
+func TestRunAgentLoop_PicoSkipsInterimPublishWhenNotAllowed(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	cfg := &config.Config{
+		Agents: config.AgentsConfig{
+			Defaults: config.AgentDefaults{
+				Workspace:         tmpDir,
+				ModelName:         "test-model",
+				MaxTokens:         4096,
+				MaxToolIterations: 10,
+			},
+		},
+	}
+
+	msgBus := bus.NewMessageBus()
+	provider := &picoInterleavedContentProvider{}
+	al := NewAgentLoop(cfg, msgBus, provider)
+
+	agent := al.GetRegistry().GetDefaultAgent()
+	if agent == nil {
+		t.Fatal("expected default agent")
+	}
+	agent.Tools.Register(&toolLimitTestTool{})
+
+	response, err := al.runAgentLoop(context.Background(), agent, processOptions{
+		SessionKey:              "agent:main:pico:session-1",
+		Channel:                 "pico",
+		ChatID:                  "session-1",
+		UserMessage:             "run with tools",
+		DefaultResponse:         defaultResponse,
+		EnableSummary:           false,
+		SendResponse:            false,
+		AllowInterimPicoPublish: false,
+		SuppressToolFeedback:    true,
+	})
+	if err != nil {
+		t.Fatalf("runAgentLoop() error = %v", err)
+	}
+	if response != "final model text" {
+		t.Fatalf("runAgentLoop() response = %q, want %q", response, "final model text")
+	}
+
+	select {
+	case outbound := <-msgBus.OutboundChan():
+		t.Fatalf("unexpected outbound message when interim publish disabled: %+v", outbound)
+	case <-time.After(200 * time.Millisecond):
+	}
+}
+
 func TestResolveMediaRefs_ResolvesToBase64(t *testing.T) {
 	store := media.NewFileMediaStore()
 	dir := t.TempDir()