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

fix(channels): send steering final replies as new messages

Browse source
This commit is contained in:
Anton Bogdanovich 2026-05-09 15:51:35 -07:00
parent 6e6293e596
commit 8fcd41a29e
11 changed files with 211 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
pkg/agent/agent.go
View file

@ -121,6 +121,7 @@ const (
messageKindThought = "thought"
messageKindToolFeedback = "tool_feedback"
messageKindToolCalls = "tool_calls"
messageKindFinalReply = "final_reply"
metadataKeyAccountID = "account_id"
metadataKeyGuildID = "guild_id"
metadataKeyTeamID = "team_id"
@ -260,7 +261,21 @@ func (al *AgentLoop) Run(ctx context.Context) error {
return
}
if continued != "" {
al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, target.SessionKey, continued)
al.publishResponseWithContextIfNeeded(
ctx,
target.Channel,
target.ChatID,
target.SessionKey,
continued,
&bus.InboundContext{
Channel: m.Context.Channel,
ChatID: m.Context.ChatID,
TopicID: m.Context.TopicID,
Raw: map[string]string{
metadataKeyMessageKind: messageKindFinalReply,
},
},
)
}
return
}
@ -554,13 +569,20 @@ func (al *AgentLoop) runAgentLoop(
opts.Dispatch.SessionKey,
opts.Dispatch.SessionScope,
)
outboundCtx := outboundContextFromInbound(
opts.Dispatch.InboundContext,
opts.Dispatch.Channel(),
opts.Dispatch.ChatID(),
opts.Dispatch.ReplyToMessageID(),
)
if result.preferNewOutboundReply {
if outboundCtx.Raw == nil {
outboundCtx.Raw = make(map[string]string, 1)
}
outboundCtx.Raw[metadataKeyMessageKind] = messageKindFinalReply
}
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Context: outboundContextFromInbound(
opts.Dispatch.InboundContext,
opts.Dispatch.Channel(),
opts.Dispatch.ChatID(),
opts.Dispatch.ReplyToMessageID(),
),
Context: outboundCtx,
AgentID: agentID,
SessionKey: sessionKey,
Scope: scope,

10
pkg/agent/agent_outbound.go
View file

@ -41,6 +41,14 @@ func (al *AgentLoop) publishResponseOrError(
}
func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatID, sessionKey, response string) {
al.publishResponseWithContextIfNeeded(ctx, channel, chatID, sessionKey, response, nil)
}
func (al *AgentLoop) publishResponseWithContextIfNeeded(
ctx context.Context,
channel, chatID, sessionKey, response string,
inboundCtx *bus.InboundContext,
) {
if response == "" {
return
}
@ -75,7 +83,7 @@ func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatI
}
msg := bus.OutboundMessage{
Context: bus.NewOutboundContext(channel, chatID, ""),
Context: outboundContextFromInbound(inboundCtx, channel, chatID, ""),
Content: response,
}
if sessionKey != "" {

16
pkg/agent/agent_steering.go
View file

@ -58,7 +58,21 @@ func (al *AgentLoop) runTurnWithSteering(ctx context.Context, initialMsg bus.Inb
// Publish final response
if finalResponse != "" {
al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, target.SessionKey, finalResponse)
al.publishResponseWithContextIfNeeded(
ctx,
target.Channel,
target.ChatID,
target.SessionKey,
finalResponse,
&bus.InboundContext{
Channel: initialMsg.Context.Channel,
ChatID: initialMsg.Context.ChatID,
TopicID: initialMsg.Context.TopicID,
Raw: map[string]string{
metadataKeyMessageKind: messageKindFinalReply,
},
},
)
}
}

3
pkg/agent/pipeline_execute.go
View file

@ -587,6 +587,7 @@ toolLoop:
}
if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 {
exec.sawSteering = true
exec.pendingMessages = append(exec.pendingMessages, steerMsgs...)
}
@ -655,6 +656,7 @@ toolLoop:
// This covers the case where tools were partially executed and skipped due to steering,
// but one tool had ResponseHandled=false (so allResponsesHandled=false).
if len(exec.pendingMessages) > 0 {
exec.sawSteering = true
logger.InfoCF("agent", "Pending steering after partial tool execution; continuing turn",
map[string]any{
"agent_id": ts.agent.ID,
@ -667,6 +669,7 @@ toolLoop:
// Poll for newly arrived steering
if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 {
exec.sawSteering = true
logger.InfoCF("agent", "Steering arrived after tool delivery; continuing turn",
map[string]any{
"agent_id": ts.agent.ID,

14
pkg/agent/pipeline_finalize.go
View file

@ -32,9 +32,10 @@ func (p *Pipeline) Finalize(
}
ts.setPhase(TurnPhaseCompleted)
return turnResult{
finalContent: finalContent,
status: turnStatus,
followUps: append([]bus.InboundMessage(nil), ts.followUps...),
finalContent: finalContent,
status: turnStatus,
followUps: append([]bus.InboundMessage(nil), ts.followUps...),
preferNewOutboundReply: exec.sawSteering,
}, nil
}
@ -75,8 +76,9 @@ func (p *Pipeline) Finalize(
ts.setPhase(TurnPhaseCompleted)
return turnResult{
finalContent: finalContent,
status: turnStatus,
followUps: append([]bus.InboundMessage(nil), ts.followUps...),
finalContent: finalContent,
status: turnStatus,
followUps: append([]bus.InboundMessage(nil), ts.followUps...),
preferNewOutboundReply: exec.sawSteering,
}, nil
}

1
pkg/agent/pipeline_llm.go
View file

@ -475,6 +475,7 @@ func (p *Pipeline) CallLLM(
responseContent = exec.response.ReasoningContent
}
if steerMsgs := al.dequeueSteeringMessagesForScope(ts.sessionKey); len(steerMsgs) > 0 {
exec.sawSteering = true
logger.InfoCF("agent", "Steering arrived after direct LLM response; continuing turn",
map[string]any{
"agent_id": ts.agent.ID,

91
pkg/agent/steering_test.go
View file

@ -851,6 +851,97 @@ func TestAgentLoop_Run_AutoContinuesLateSteeringMessage(t *testing.T) {
}
}
func TestAgentLoop_RunTurnWithSteering_PublishesFinalReplyAsNewMessage(t *testing.T) {
tmpDir, err := os.MkdirTemp("", "agent-test-*")
if err != nil {
t.Fatalf("Failed to create temp dir: %v", err)
}
defer os.RemoveAll(tmpDir)
cfg := &config.Config{
Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{
Workspace: tmpDir,
ModelName: "test-model",
MaxTokens: 4096,
MaxToolIterations: 10,
},
},
}
sessionKey := session.BuildMainSessionKey(routing.DefaultAgentID)
provider := &blockingDirectProvider{
firstStarted: make(chan struct{}),
releaseFirst: make(chan struct{}),
firstResp: "stale direct response",
finalResp: "fresh response after steering",
}
msgBus := bus.NewMessageBus()
al := NewAgentLoop(cfg, msgBus, provider)
msg := testInboundMessage(bus.InboundMessage{
Context: bus.InboundContext{
Channel: "telegram",
ChatID: "-1001234567890",
ChatType: "group",
TopicID: "6",
SenderID: "user-1",
MessageID: "475",
},
SessionKey: sessionKey,
Content: "initial request",
})
done := make(chan struct{})
go func() {
al.runTurnWithSteering(context.Background(), msg)
close(done)
}()
select {
case <-provider.firstStarted:
case <-time.After(2 * time.Second):
t.Fatal("timeout waiting for first LLM call to start")
}
if err := al.Steer(providers.Message{Role: "user", Content: "follow-up instruction"}); err != nil {
t.Fatalf("Steer failed: %v", err)
}
close(provider.releaseFirst)
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("timeout waiting for steering turn to finish")
}
var finalOutbound bus.OutboundMessage
found := false
drain:
for {
select {
case outbound := <-msgBus.OutboundChan():
if outbound.Content == "fresh response after steering" {
finalOutbound = outbound
found = true
}
default:
break drain
}
}
if !found {
t.Fatal("expected final outbound response")
}
if got := finalOutbound.Context.Raw[metadataKeyMessageKind]; got != messageKindFinalReply {
t.Fatalf("message kind = %q, want %q", got, messageKindFinalReply)
}
if finalOutbound.Context.TopicID != "6" {
t.Fatalf("topic_id = %q, want 6", finalOutbound.Context.TopicID)
}
}
func TestAgentLoop_Run_QueuedVoiceMessageIsTranscribedBeforeSteering(t *testing.T) {
tmpDir := t.TempDir()
cfg := &config.Config{

2
pkg/agent/turn_coord.go
View file

@ -89,11 +89,13 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState, pipeline *Pipel
// We do NOT call dequeueSteeringMessagesForScope here because
// steering was already consumed from al.steering by ExecuteTools.
if len(exec.pendingMessages) > 0 {
exec.sawSteering = true
pendingMessages = append(pendingMessages, exec.pendingMessages...)
exec.pendingMessages = nil
}
} else if !ts.opts.SkipInitialSteeringPoll {
if steerMsgs := al.dequeueSteeringMessagesForScopeWithFallback(ts.sessionKey); len(steerMsgs) > 0 {
exec.sawSteering = true
pendingMessages = append(pendingMessages, steerMsgs...)
}
}

8
pkg/agent/turn_state.go
View file

@ -81,9 +81,10 @@ const (
// =============================================================================
type turnResult struct {
finalContent string
status TurnEndStatus
followUps []bus.InboundMessage
finalContent string
status TurnEndStatus
followUps []bus.InboundMessage
preferNewOutboundReply bool
}
// =============================================================================
@ -118,6 +119,7 @@ type turnExecution struct {
// Turn output
finalContent string
sawSteering bool
// Iteration tracking
iteration int

7
pkg/channels/manager.go
View file

@ -167,7 +167,9 @@ func outboundMessageBypassesPlaceholderEdit(msg bus.OutboundMessage) bool {
return false
}
kind := strings.TrimSpace(msg.Context.Raw["message_kind"])
return strings.EqualFold(kind, "thought") || strings.EqualFold(kind, "tool_calls")
return strings.EqualFold(kind, "thought") ||
strings.EqualFold(kind, "tool_calls") ||
strings.EqualFold(kind, "final_reply")
}
func outboundMediaChannel(msg bus.OutboundMediaMessage) string {
@ -387,6 +389,9 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess
if deleter, ok := ch.(MessageDeleter); ok {
deleter.DeleteMessage(ctx, chatID, entry.id) // best effort
}
if strings.EqualFold(strings.TrimSpace(msg.Context.Raw["message_kind"]), "final_reply") {
dismissTrackedToolFeedbackMessage(ctx, ch, chatID, &msg.Context)
}
return nil, false
}
if editor, ok := ch.(MessageEditor); ok {

42
pkg/channels/manager_test.go
View file

@ -1306,6 +1306,48 @@ func TestPreSend_ThoughtPlaceholderDeleteAndSkipsEdit(t *testing.T) {
}
}
func TestPreSend_FinalReplyPlaceholderDeleteAndDismissesTrackedFeedback(t *testing.T) {
m := newTestManager()
ch := &mockDeletingMessageEditor{
mockMessageEditor: mockMessageEditor{
editFn: func(_ context.Context, _, _, _ string) error {
t.Fatal("expected final reply to bypass placeholder edit")
return nil
},
},
}
m.RecordPlaceholder("test", "123", "456")
msg := testOutboundMessage(bus.OutboundMessage{
Channel: "test",
ChatID: "123",
Content: "final aggregated reply",
Context: bus.InboundContext{
Channel: "test",
ChatID: "123",
Raw: map[string]string{
"message_kind": "final_reply",
},
},
})
msgIDs, handled := m.preSend(context.Background(), "test", msg, ch)
if handled {
t.Fatalf("expected preSend to fall through so the channel can send a new final reply, got %v", msgIDs)
}
if ch.deleteCalls != 1 {
t.Fatalf("expected placeholder deletion, got %d delete calls", ch.deleteCalls)
}
if ch.deletedChatID != "123" || ch.deletedMessageID != "456" {
t.Fatalf("unexpected placeholder deletion target: %s/%s", ch.deletedChatID, ch.deletedMessageID)
}
if ch.dismissedChatID != "123" {
t.Fatalf("expected tracked tool feedback dismissal, got %q", ch.dismissedChatID)
}
}
func TestSendWithRetry_ToolCallsPlaceholderDeleteAndFallsThroughToSend(t *testing.T) {
m := newTestManager()