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

feat(session): persist threaded message metadata

Browse source
This commit is contained in:
Dmitrii Balabanov 2026-03-31 14:46:24 +03:00
parent b114dcaeb1
commit 6ea3ce6aa0
12 changed files with 606 additions and 172 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
pkg/agent/eventbus_test.go
View file

@ -140,8 +140,8 @@ func TestAgentLoop_EmitsMinimalTurnEvents(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if response != "done" {
t.Fatalf("expected final response 'done', got %q", response)
if response.Content != "done" {
t.Fatalf("expected final response 'done', got %q", response.Content)
}
events := collectEventStream(sub.C)
@ -396,8 +396,8 @@ func TestAgentLoop_EmitsContextCompressEventOnRetry(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "Recovered from context error" {
t.Fatalf("expected retry success, got %q", resp)
if resp.Content != "Recovered from context error" {
t.Fatalf("expected retry success, got %q", resp.Content)
}
events := collectEventStream(sub.C)
@ -552,8 +552,8 @@ func TestAgentLoop_EmitsFollowUpQueuedEvent(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "async launched" {
t.Fatalf("expected final response 'async launched', got %q", resp)
if resp.Content != "async launched" {
t.Fatalf("expected final response 'async launched', got %q", resp.Content)
}
select {

12
pkg/agent/hook_process_test.go
View file

@ -52,8 +52,8 @@ func TestAgentLoop_MountProcessHook_LLMAndObserver(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "provider content|ipc" {
t.Fatalf("expected process-hooked llm content, got %q", resp)
if resp.Content != "provider content|ipc" {
t.Fatalf("expected process-hooked llm content, got %q", resp.Content)
}
provider.mu.Lock()
@ -92,8 +92,8 @@ func TestAgentLoop_MountProcessHook_ToolRewrite(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "ipc:ipc" {
t.Fatalf("expected rewritten process-hook tool result, got %q", resp)
if resp.Content != "ipc:ipc" {
t.Fatalf("expected rewritten process-hook tool result, got %q", resp.Content)
}
}
@ -160,8 +160,8 @@ func TestAgentLoop_MountProcessHook_ApprovalDeny(t *testing.T) {
}
expected := "Tool execution denied by approval hook: blocked by ipc hook"
if resp != expected {
t.Fatalf("expected %q, got %q", expected, resp)
if resp.Content != expected {
t.Fatalf("expected %q, got %q", expected, resp.Content)
}
events := collectEventStream(sub.C)

12
pkg/agent/hooks_test.go
View file

@ -159,8 +159,8 @@ func TestAgentLoop_Hooks_ObserverAndLLMInterceptor(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "hooked content" {
t.Fatalf("expected hooked content, got %q", resp)
if resp.Content != "hooked content" {
t.Fatalf("expected hooked content, got %q", resp.Content)
}
provider.mu.Lock()
@ -286,8 +286,8 @@ func TestAgentLoop_Hooks_ToolInterceptorCanRewrite(t *testing.T) {
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "after:modified" {
t.Fatalf("expected rewritten tool result, got %q", resp)
if resp.Content != "after:modified" {
t.Fatalf("expected rewritten tool result, got %q", resp.Content)
}
}
@ -326,8 +326,8 @@ func TestAgentLoop_Hooks_ToolApproverCanDeny(t *testing.T) {
t.Fatalf("runAgentLoop failed: %v", err)
}
expected := "Tool execution denied by approval hook: blocked"
if resp != expected {
t.Fatalf("expected %q, got %q", expected, resp)
if resp.Content != expected {
t.Fatalf("expected %q, got %q", expected, resp.Content)
}
events := collectEventStream(sub.C)

326
pkg/agent/loop.go
View file

@ -73,24 +73,26 @@ type AgentLoop struct {
// processOptions configures how a message is processed
type processOptions struct {
SessionKey string // Session identifier for history/context
Channel string // Target channel for tool execution
ChatID string // Target chat ID for tool execution
MessageID string // Current inbound platform message ID
ReplyToMessageID string // Current inbound reply target message ID
SenderID string // Current sender ID for dynamic context
SenderDisplayName string // Current sender display name for dynamic context
UserMessage string // User message content (may include prefix)
ForcedSkills []string // Skills explicitly requested for this message
SystemPromptOverride string // Override the default system prompt (Used by SubTurns)
Media []string // media:// refs from inbound message
InitialSteeringMessages []providers.Message // Steering messages from refactor/agent
DefaultResponse string // Response when LLM returns empty
EnableSummary bool // Whether to trigger summarization
SendResponse bool // Whether to send response via bus
SuppressToolFeedback bool // Whether to suppress inline tool feedback messages
NoHistory bool // If true, don't load session history (for heartbeat)
SkipInitialSteeringPoll bool // If true, skip the steering poll at loop start (used by Continue)
SessionKey string // Session identifier for history/context
Channel string // Target channel for tool execution
ChatID string // Target chat ID for tool execution
MessageID string // Current inbound platform message ID
ReplyToMessageID string // Current inbound reply target message ID
SenderID string // Current sender ID for dynamic context
SenderDisplayName string // Current sender display name for dynamic context
UserMessage string // User message content (may include prefix)
ForcedSkills []string // Skills explicitly requested for this message
SystemPromptOverride string // Override the default system prompt (Used by SubTurns)
Media []string // media:// refs from inbound message
InitialSteeringMessages []providers.Message // Steering messages from refactor/agent
EphemeralPrefix []providers.Message // Messages prepended to LLM context but NOT saved to history
DefaultResponse string // Response when LLM returns empty
EnableSummary bool // Whether to trigger summarization
SendResponse bool // Whether to send response via bus
SuppressToolFeedback bool // Whether to suppress inline tool feedback messages
NoHistory bool // If true, don't load session history (for heartbeat)
SkipInitialSteeringPoll bool // If true, skip the steering poll at loop start (used by Continue)
Sender *providers.MessageSender // Author identity (nil for system/automated messages)
}
type continuationTarget struct {
@ -99,6 +101,46 @@ type continuationTarget struct {
ChatID string
}
type agentResponse struct {
Content string
Channel string
ChatID string
OnDelivered func(msgIDs []string)
}
func (r agentResponse) outboundMessage(defaultChannel, defaultChatID string) bus.OutboundMessage {
channel := r.Channel
if channel == "" {
channel = defaultChannel
}
chatID := r.ChatID
if chatID == "" {
chatID = defaultChatID
}
return bus.OutboundMessage{
Channel: channel,
ChatID: chatID,
Content: r.Content,
OnDelivered: r.OnDelivered,
}
}
func singleMessageIDs(msgID string) []string {
if msgID == "" {
return nil
}
return []string{msgID}
}
func cloneMessageIDs(msgIDs []string) []string {
if len(msgIDs) == 0 {
return nil
}
cloned := make([]string, len(msgIDs))
copy(cloned, msgIDs)
return cloned
}
const (
defaultResponse = "The model returned an empty response. This may indicate a provider error or token limit."
toolLimitResponse = "I've reached `max_tool_iterations` without a final response. Increase `max_tool_iterations` in config.json if this task needs more tool steps."
@ -508,9 +550,12 @@ func (al *AgentLoop) Run(ctx context.Context) error {
response, err := al.processMessage(ctx, msg)
if err != nil {
response = fmt.Sprintf("Error processing message: %v", err)
response = agentResponse{
Content: fmt.Sprintf("Error processing message: %v", err),
Channel: msg.Channel,
ChatID: msg.ChatID,
}
}
finalResponse := response
target, targetErr := al.buildContinuationTarget(msg)
if targetErr != nil {
@ -523,12 +568,17 @@ func (al *AgentLoop) Run(ctx context.Context) error {
}
if target == nil {
cancelDrain()
if finalResponse != "" {
al.PublishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, finalResponse)
if response.Content != "" {
al.publishAgentResponseIfNeeded(ctx, response, msg.Channel, msg.ChatID)
}
return
}
if response.Content != "" {
al.publishAgentResponseIfNeeded(ctx, response, target.Channel, target.ChatID)
}
prevContent := response.Content
for al.pendingSteeringCountForScope(target.SessionKey) > 0 {
logger.InfoCF("agent", "Continuing queued steering after turn end",
map[string]any{
@ -538,7 +588,13 @@ func (al *AgentLoop) Run(ctx context.Context) error {
"queue_depth": al.pendingSteeringCountForScope(target.SessionKey),
})
continued, continueErr := al.Continue(ctx, target.SessionKey, target.Channel, target.ChatID)
continued, continueErr := al.continueResponse(
ctx,
target.SessionKey,
target.Channel,
target.ChatID,
prevContent,
)
if continueErr != nil {
logger.WarnCF("agent", "Failed to continue queued steering",
map[string]any{
@ -548,11 +604,11 @@ func (al *AgentLoop) Run(ctx context.Context) error {
})
return
}
if continued == "" {
if continued.Content == "" {
return
}
finalResponse = continued
al.publishAgentResponseIfNeeded(ctx, continued, target.Channel, target.ChatID)
prevContent = continued.Content
}
cancelDrain()
@ -566,7 +622,13 @@ func (al *AgentLoop) Run(ctx context.Context) error {
"queue_depth": al.pendingSteeringCountForScope(target.SessionKey),
})
continued, continueErr := al.Continue(ctx, target.SessionKey, target.Channel, target.ChatID)
continued, continueErr := al.continueResponse(
ctx,
target.SessionKey,
target.Channel,
target.ChatID,
prevContent,
)
if continueErr != nil {
logger.WarnCF("agent", "Failed to continue queued steering after shutdown drain",
map[string]any{
@ -576,15 +638,11 @@ func (al *AgentLoop) Run(ctx context.Context) error {
})
return
}
if continued == "" {
if continued.Content == "" {
break
}
finalResponse = continued
}
if finalResponse != "" {
al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse)
al.publishAgentResponseIfNeeded(ctx, continued, target.Channel, target.ChatID)
prevContent = continued.Content
}
}()
}
@ -648,10 +706,17 @@ func (al *AgentLoop) drainBusToSteering(ctx context.Context, activeScope, active
"scope": activeScope,
})
replyToMessageID := msg.ReplyToMessageID
if replyToMessageID == "" {
replyToMessageID = inboundMetadata(msg, metadataKeyReplyToMessage)
}
if err := al.enqueueSteeringMessage(activeScope, activeAgentID, providers.Message{
Role: "user",
Content: msg.Content,
Media: append([]string(nil), msg.Media...),
Role: "user",
Content: msg.Content,
Media: append([]string(nil), msg.Media...),
MessageIDs: singleMessageIDs(msg.MessageID),
ReplyToMessageID: replyToMessageID,
Sender: messageSenderFromInbound(msg.Sender),
}); err != nil {
logger.WarnCF("agent", "Failed to steer message, will be lost",
map[string]any{
@ -666,8 +731,22 @@ func (al *AgentLoop) Stop() {
al.running.Store(false)
}
// PublishResponseIfNeeded is the public adapter that satisfies the tools.JobExecutor interface.
func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatID, response string) {
if response == "" {
al.publishAgentResponseIfNeeded(
ctx,
agentResponse{Content: response, Channel: channel, ChatID: chatID},
channel,
chatID,
)
}
func (al *AgentLoop) publishAgentResponseIfNeeded(
ctx context.Context,
response agentResponse,
defaultChannel, defaultChatID string,
) {
if response.Content == "" {
return
}
@ -685,21 +764,18 @@ func (al *AgentLoop) PublishResponseIfNeeded(ctx context.Context, channel, chatI
logger.DebugCF(
"agent",
"Skipped outbound (message tool already sent)",
map[string]any{"channel": channel},
map[string]any{"channel": response.outboundMessage(defaultChannel, defaultChatID).Channel},
)
return
}
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: channel,
ChatID: chatID,
Content: response,
})
outbound := response.outboundMessage(defaultChannel, defaultChatID)
al.bus.PublishOutbound(ctx, outbound)
logger.InfoCF("agent", "Published outbound response",
map[string]any{
"channel": channel,
"chat_id": chatID,
"content_len": len(response),
"channel": outbound.Channel,
"chat_id": outbound.ChatID,
"content_len": len(response.Content),
})
}
@ -1305,7 +1381,16 @@ func (al *AgentLoop) ProcessDirectWithChannel(
SessionKey: sessionKey,
}
return al.processMessage(ctx, msg)
response, err := al.processMessage(ctx, msg)
if err != nil {
return "", err
}
// ProcessDirectWithChannel is synchronous: the caller receives the content
// directly, so the response is considered "delivered" immediately.
if response.OnDelivered != nil {
response.OnDelivered(nil)
}
return response.Content, nil
}
// ProcessHeartbeat processes a heartbeat request without session history.
@ -1325,7 +1410,7 @@ func (al *AgentLoop) ProcessHeartbeat(
if agent == nil {
return "", fmt.Errorf("no default agent for heartbeat")
}
return al.runAgentLoop(ctx, agent, processOptions{
response, err := al.runAgentLoop(ctx, agent, processOptions{
SessionKey: "heartbeat",
Channel: channel,
ChatID: chatID,
@ -1336,9 +1421,13 @@ func (al *AgentLoop) ProcessHeartbeat(
SuppressToolFeedback: true,
NoHistory: true, // Don't load session history for heartbeat
})
if err != nil {
return "", err
}
return response.Content, nil
}
func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (agentResponse, error) {
// Add message preview to log (show full content for error messages)
var logContent string
if strings.Contains(msg.Content, "Error:") || strings.Contains(msg.Content, "error") {
@ -1373,7 +1462,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
route, agent, routeErr := al.resolveMessageRoute(msg)
if routeErr != nil {
return "", routeErr
return agentResponse{}, routeErr
}
// Reset message-tool state for this round so we don't skip publishing due to a previous round.
@ -1402,20 +1491,28 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
Channel: msg.Channel,
ChatID: msg.ChatID,
MessageID: msg.MessageID,
ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage),
ReplyToMessageID: msg.ReplyToMessageID,
SenderID: msg.SenderID,
SenderDisplayName: msg.Sender.DisplayName,
Sender: messageSenderFromInbound(msg.Sender),
UserMessage: msg.Content,
Media: msg.Media,
DefaultResponse: defaultResponse,
EnableSummary: true,
SendResponse: false,
}
if opts.ReplyToMessageID == "" {
opts.ReplyToMessageID = inboundMetadata(msg, metadataKeyReplyToMessage)
}
// context-dependent commands check their own Runtime fields and report
// "unavailable" when the required capability is nil.
if response, handled := al.handleCommand(ctx, msg, agent, &opts); handled {
return response, nil
return agentResponse{
Content: response,
Channel: opts.Channel,
ChatID: opts.ChatID,
}, nil
}
if pending := al.takePendingSkills(opts.SessionKey); len(pending) > 0 {
@ -1488,9 +1585,9 @@ func (al *AgentLoop) requeueInboundMessage(msg bus.InboundMessage) error {
func (al *AgentLoop) processSystemMessage(
ctx context.Context,
msg bus.InboundMessage,
) (string, error) {
) (agentResponse, error) {
if msg.Channel != "system" {
return "", fmt.Errorf(
return agentResponse{}, fmt.Errorf(
"processSystemMessage called with non-system message channel: %s",
msg.Channel,
)
@ -1527,13 +1624,13 @@ func (al *AgentLoop) processSystemMessage(
"content_len": len(content),
"channel": originChannel,
})
return "", nil
return agentResponse{}, nil
}
// Use default agent for system messages
agent := al.GetRegistry().GetDefaultAgent()
if agent == nil {
return "", fmt.Errorf("no default agent for system message")
return agentResponse{}, fmt.Errorf("no default agent for system message")
}
// Use the origin session for context
@ -1556,7 +1653,7 @@ func (al *AgentLoop) runAgentLoop(
ctx context.Context,
agent *AgentInstance,
opts processOptions,
) (string, error) {
) (agentResponse, error) {
// Record last channel for heartbeat notifications (skip internal channels and cli)
if opts.Channel != "" && opts.ChatID != "" && !constants.IsInternalChannel(opts.Channel) {
channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID)
@ -1572,10 +1669,10 @@ func (al *AgentLoop) runAgentLoop(
ts := newTurnState(agent, opts, al.newTurnEventScope(agent.ID, opts.SessionKey))
result, err := al.runTurn(ctx, ts)
if err != nil {
return "", err
return agentResponse{}, err
}
if result.status == TurnEndStatusAborted {
return "", nil
return agentResponse{}, nil
}
for _, followUp := range result.followUps {
@ -1588,12 +1685,35 @@ func (al *AgentLoop) runAgentLoop(
}
}
if opts.SendResponse && result.finalContent != "" {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
Content: result.finalContent,
})
response := agentResponse{
Content: result.finalContent,
Channel: opts.Channel,
ChatID: opts.ChatID,
}
if !opts.NoHistory && result.finalContent != "" {
var once sync.Once
response.OnDelivered = func(msgIDs []string) {
once.Do(func() {
assistantMsg := providers.Message{
Role: "assistant",
Content: result.finalContent,
MessageIDs: cloneMessageIDs(msgIDs),
}
agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg)
if saveErr := agent.Sessions.Save(opts.SessionKey); saveErr != nil {
logger.WarnCF("agent", "Failed to save delivered assistant message",
map[string]any{
"session_key": opts.SessionKey,
"error": saveErr.Error(),
})
return
}
if opts.EnableSummary {
al.maybeSummarize(agent, opts.SessionKey, ts.scope)
}
})
}
}
if result.finalContent != "" {
@ -1607,7 +1727,7 @@ func (al *AgentLoop) runAgentLoop(
})
}
return result.finalContent, nil
return response, nil
}
func (al *AgentLoop) targetReasoningChannelID(channelName string) (chatID string) {
@ -1771,15 +1891,14 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er
// Save user message to session (from Incoming)
if !ts.opts.NoHistory && (strings.TrimSpace(ts.userMessage) != "" || len(ts.media) > 0) {
rootMsg := providers.Message{
Role: "user",
Content: ts.userMessage,
Media: append([]string(nil), ts.media...),
}
if len(rootMsg.Media) > 0 {
ts.agent.Sessions.AddFullMessage(ts.sessionKey, rootMsg)
} else {
ts.agent.Sessions.AddMessage(ts.sessionKey, rootMsg.Role, rootMsg.Content)
Role: "user",
Content: ts.userMessage,
Media: append([]string(nil), ts.media...),
MessageIDs: singleMessageIDs(ts.opts.MessageID),
ReplyToMessageID: ts.opts.ReplyToMessageID,
Sender: ts.opts.Sender,
}
ts.agent.Sessions.AddFullMessage(ts.sessionKey, rootMsg)
ts.recordPersistedMessage(rootMsg)
ts.ingestMessage(turnCtx, al, rootMsg)
}
@ -1790,6 +1909,12 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er
activeProvider = ts.agent.LightProvider
}
pendingMessages := append([]providers.Message(nil), ts.opts.InitialSteeringMessages...)
// Inject ephemeral prefix into LLM context without saving to session history.
// Used to carry the previous turn's assistant reply into a steering continuation
// so the model sees it even before OnDelivered persists it.
if len(ts.opts.EphemeralPrefix) > 0 {
messages = append(messages, ts.opts.EphemeralPrefix...)
}
var finalContent string
turnLoop:
@ -2214,8 +2339,10 @@ turnLoop:
if reasoningContent == "" {
reasoningContent = response.ReasoningContent
}
// Use the parent ctx, not turnCtx: reasoning is fire-and-forget and
// must not be gated by the turn's own cancel (which fires on return).
go al.handleReasoning(
turnCtx,
ctx,
reasoningContent,
ts.channel,
al.targetReasoningChannelID(ts.channel),
@ -2824,34 +2951,6 @@ turnLoop:
ts.setPhase(TurnPhaseFinalizing)
ts.setFinalContent(finalContent)
if !ts.opts.NoHistory {
finalMsg := providers.Message{Role: "assistant", Content: finalContent}
ts.agent.Sessions.AddMessage(ts.sessionKey, finalMsg.Role, finalMsg.Content)
ts.recordPersistedMessage(finalMsg)
ts.ingestMessage(turnCtx, al, finalMsg)
if err := ts.agent.Sessions.Save(ts.sessionKey); err != nil {
turnStatus = TurnEndStatusError
al.emitEvent(
EventKindError,
ts.eventMeta("runTurn", "turn.error"),
ErrorPayload{
Stage: "session_save",
Message: err.Error(),
},
)
return turnResult{}, err
}
}
if ts.opts.EnableSummary {
al.contextManager.Compact(
turnCtx,
&CompactRequest{
SessionKey: ts.sessionKey,
Reason: ContextCompressReasonSummarize,
},
)
}
ts.setPhase(TurnPhaseCompleted)
return turnResult{
@ -3423,3 +3522,22 @@ func extractProvider(registry *AgentRegistry) (providers.LLMProvider, bool) {
}
return defaultAgent.Provider, true
}
// messageSenderFromInbound converts bus.SenderInfo to providers.MessageSender.
// Returns nil if no meaningful identity is present.
func messageSenderFromInbound(s bus.SenderInfo) *providers.MessageSender {
if s.Username == "" && s.FirstName == "" && s.LastName == "" && s.DisplayName == "" {
return nil
}
username := s.Username
firstName := s.FirstName
lastName := s.LastName
if firstName == "" && lastName == "" && s.DisplayName != "" {
firstName = s.DisplayName
}
return &providers.MessageSender{
Username: username,
FirstName: firstName,
LastName: lastName,
}
}

153
pkg/agent/loop_test.go
View file

@ -151,8 +151,8 @@ func TestProcessMessage_IncludesCurrentSenderInDynamicContext(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "Mock response" {
t.Fatalf("processMessage() response = %q, want %q", response, "Mock response")
if response.Content != "Mock response" {
t.Fatalf("processMessage() response = %q, want %q", response.Content, "Mock response")
}
if len(provider.lastMessages) == 0 {
t.Fatal("provider did not receive any messages")
@ -207,8 +207,8 @@ func TestProcessMessage_UseCommandLoadsRequestedSkill(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "Mock response" {
t.Fatalf("processMessage() response = %q, want %q", response, "Mock response")
if response.Content != "Mock response" {
t.Fatalf("processMessage() response = %q, want %q", response.Content, "Mock response")
}
if len(provider.lastMessages) == 0 {
t.Fatal("provider did not receive any messages")
@ -297,8 +297,8 @@ func TestProcessMessage_UseCommandArmsSkillForNextMessage(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() arm error = %v", err)
}
if !strings.Contains(response, `Skill "shell" is armed for your next message.`) {
t.Fatalf("arm response = %q, want armed confirmation", response)
if !strings.Contains(response.Content, `Skill "shell" is armed for your next message.`) {
t.Fatalf("arm response = %q, want armed confirmation", response.Content)
}
response, err = al.processMessage(context.Background(), bus.InboundMessage{
@ -310,8 +310,8 @@ func TestProcessMessage_UseCommandArmsSkillForNextMessage(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() follow-up error = %v", err)
}
if response != "Mock response" {
t.Fatalf("follow-up response = %q, want %q", response, "Mock response")
if response.Content != "Mock response" {
t.Fatalf("follow-up response = %q, want %q", response.Content, "Mock response")
}
if len(provider.lastMessages) == 0 {
t.Fatal("provider did not receive any messages")
@ -628,8 +628,8 @@ func TestProcessMessage_MediaToolHandledSkipsFollowUpLLMAndFinalText(t *testing.
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "" {
t.Fatalf("expected no final response when media tool already handled delivery, got %q", response)
if response.Content != "" {
t.Fatalf("expected no final response when media tool already handled delivery, got %q", response.Content)
}
if provider.calls != 1 {
t.Fatalf("expected exactly 1 LLM call, got %d", provider.calls)
@ -723,8 +723,8 @@ func TestProcessMessage_HandledToolProcessesQueuedSteeringBeforeReturning(t *tes
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "Handled the queued steering message." {
t.Fatalf("response = %q, want queued steering response", response)
if response.Content != "Handled the queued steering message." {
t.Fatalf("response = %q, want queued steering response", response.Content)
}
if provider.calls != 2 {
t.Fatalf("expected 2 LLM calls after queued steering, got %d", provider.calls)
@ -774,8 +774,8 @@ func TestProcessMessage_MediaArtifactCanBeForwardedBySendFile(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "" {
t.Fatalf("expected no final response after send_file handled delivery, got %q", response)
if response.Content != "" {
t.Fatalf("expected no final response after send_file handled delivery, got %q", response.Content)
}
if provider.calls != 2 {
t.Fatalf("expected 2 LLM calls (artifact + send_file), got %d", provider.calls)
@ -1361,7 +1361,13 @@ func (h testHelper) executeAndGetResponse(tb testing.TB, ctx context.Context, ms
if err != nil {
tb.Fatalf("processMessage failed: %v", err)
}
return response
// In tests there is no channel manager to deliver the message and call
// OnDelivered with real platform IDs. Call it with nil to trigger the
// session-history save synchronously, simulating a successful delivery.
if response.OnDelivered != nil {
response.OnDelivered(nil)
}
return response.Content
}
const responseTimeout = 3 * time.Second
@ -2448,8 +2454,8 @@ func TestProcessMessage_PublishesReasoningContentToReasoningChannel(t *testing.T
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "final answer" {
t.Fatalf("processMessage() response = %q, want %q", response, "final answer")
if response.Content != "final answer" {
t.Fatalf("processMessage() response = %q, want %q", response.Content, "final answer")
}
select {
@ -2554,8 +2560,8 @@ func TestProcessMessage_PublishesToolFeedbackWhenEnabled(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "HEARTBEAT_OK" {
t.Fatalf("processMessage() response = %q, want %q", response, "HEARTBEAT_OK")
if response.Content != "HEARTBEAT_OK" {
t.Fatalf("processMessage() response = %q, want %q", response.Content, "HEARTBEAT_OK")
}
select {
@ -2998,8 +3004,8 @@ func TestProcessMessage_ContextOverflowRecovery(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "Recovered from overflow" {
t.Fatalf("response = %q, want %q", response, "Recovered from overflow")
if response.Content != "Recovered from overflow" {
t.Fatalf("response = %q, want %q", response.Content, "Recovered from overflow")
}
if provider.calls != 2 {
@ -3039,10 +3045,111 @@ func TestProcessMessage_ContextOverflow_AnthropicStyle(t *testing.T) {
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if !strings.Contains(response, "Anthropic recovery success") {
t.Fatalf("response = %q, want success message", response)
if !strings.Contains(response.Content, "Anthropic recovery success") {
t.Fatalf("response = %q, want success message", response.Content)
}
if provider.calls != 2 {
t.Fatalf("expected 2 calls for retry, got %d", provider.calls)
}
}
// TestOnDelivered_SavesAssistantMessageWithRealIDs verifies the two-phase
// persistence design: the assistant message is absent from session history until
// OnDelivered fires, and the MessageIDs passed to the callback are recorded.
func TestOnDelivered_SavesAssistantMessageWithRealIDs(t *testing.T) {
tmpDir, err := os.MkdirTemp("", "agent-test-*")
if err != nil {
t.Fatalf("MkdirTemp: %v", err)
}
defer os.RemoveAll(tmpDir)
cfg := &config.Config{
Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{
Workspace: tmpDir,
ModelName: "test-model",
MaxTokens: 4096,
MaxToolIterations: 10,
},
},
}
msgBus := bus.NewMessageBus()
provider := &simpleMockProvider{response: "hello world"}
al := NewAgentLoop(cfg, msgBus, provider)
ctx := context.Background()
// ProcessDirectWithChannel calls OnDelivered(nil) internally, so call
// processMessage directly to control when delivery is confirmed.
msg := bus.InboundMessage{
Channel: "test",
SenderID: "cron",
ChatID: "chat1",
Content: "say hello",
}
response, err := al.processMessage(ctx, msg)
if err != nil {
t.Fatalf("processMessage: %v", err)
}
if response.Content != "hello world" {
t.Fatalf("content = %q, want %q", response.Content, "hello world")
}
if response.OnDelivered == nil {
t.Fatal("OnDelivered must be set for a non-empty response")
}
// Resolve the session key the same way processMessage did.
agent := al.registry.GetDefaultAgent()
route := al.registry.ResolveRoute(routing.RouteInput{
Channel: "test",
Peer: &routing.RoutePeer{Kind: "direct", ID: "cron"},
})
sessionKey := route.SessionKey
// Before OnDelivered fires: only the user message is in history.
histBefore := agent.Sessions.GetHistory(sessionKey)
for _, m := range histBefore {
if m.Role == "assistant" {
t.Fatalf("assistant message in history before OnDelivered: %+v", m)
}
}
// Simulate the channel manager confirming delivery with two chunk IDs
// (e.g. a long message split into two Telegram messages).
wantIDs := []string{"platform-chunk-1", "platform-chunk-2"}
response.OnDelivered(wantIDs)
// After OnDelivered fires: the assistant message must appear with the real IDs.
histAfter := agent.Sessions.GetHistory(sessionKey)
var got *providers.Message
for i := range histAfter {
if histAfter[i].Role == "assistant" {
got = &histAfter[i]
break
}
}
if got == nil {
t.Fatal("assistant message absent from history after OnDelivered")
}
if got.Content != "hello world" {
t.Fatalf("assistant content = %q, want %q", got.Content, "hello world")
}
if !slices.Equal(got.MessageIDs, wantIDs) {
t.Fatalf("MessageIDs = %v, want %v", got.MessageIDs, wantIDs)
}
// Calling OnDelivered a second time (idempotency via sync.Once) must not
// duplicate the assistant message in history.
response.OnDelivered([]string{"should-be-ignored"})
histFinal := agent.Sessions.GetHistory(sessionKey)
assistantCount := 0
for _, m := range histFinal {
if m.Role == "assistant" {
assistantCount++
}
}
if assistantCount != 1 {
t.Fatalf("expected 1 assistant message after duplicate OnDelivered, got %d", assistantCount)
}
}

39
pkg/agent/steering.go
View file

@ -290,8 +290,16 @@ func (al *AgentLoop) continueWithSteeringMessages(
ctx context.Context,
agent *AgentInstance,
sessionKey, channel, chatID string,
prevContent string,
steeringMsgs []providers.Message,
) (string, error) {
) (agentResponse, error) {
// Pass the previous assistant reply as ephemeral context so the model can
// see it without it being persisted to session history prematurely (that
// happens asynchronously via OnDelivered after channel delivery).
var ephemeral []providers.Message
if prevContent != "" {
ephemeral = []providers.Message{{Role: "assistant", Content: prevContent}}
}
return al.runAgentLoop(ctx, agent, processOptions{
SessionKey: sessionKey,
Channel: channel,
@ -299,6 +307,7 @@ func (al *AgentLoop) continueWithSteeringMessages(
DefaultResponse: defaultResponse,
EnableSummary: true,
SendResponse: false,
EphemeralPrefix: ephemeral,
InitialSteeringMessages: steeringMsgs,
SkipInitialSteeringPoll: true,
})
@ -319,31 +328,33 @@ func (al *AgentLoop) agentForSession(sessionKey string) *AgentInstance {
return registry.GetDefaultAgent()
}
// Continue resumes an idle agent by dequeuing any pending steering messages
// and running them through the agent loop. This is used when the agent's last
// message was from the assistant (i.e., it has stopped processing) and the
// user has since enqueued steering messages.
//
// If no steering messages are pending, it returns an empty string.
func (al *AgentLoop) Continue(ctx context.Context, sessionKey, channel, chatID string) (string, error) {
// continueResponse dequeues pending steering messages and runs them through the agent loop.
// prevContent is the assistant reply from the immediately preceding turn; it is injected as
// ephemeral context so the model can reference it before OnDelivered persists it to history.
// Returns an agentResponse with OnDelivered set for delayed session persistence.
// If no steering messages are pending, returns an empty agentResponse.
func (al *AgentLoop) continueResponse(
ctx context.Context,
sessionKey, channel, chatID, prevContent string,
) (agentResponse, error) {
if active := al.GetActiveTurn(); active != nil {
return "", fmt.Errorf("turn %s is still active", active.TurnID)
return agentResponse{}, fmt.Errorf("turn %s is still active", active.TurnID)
}
if err := al.ensureHooksInitialized(ctx); err != nil {
return "", err
return agentResponse{}, err
}
if err := al.ensureMCPInitialized(ctx); err != nil {
return "", err
return agentResponse{}, err
}
steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(sessionKey)
if len(steeringMsgs) == 0 {
return "", nil
return agentResponse{}, nil
}
agent := al.agentForSession(sessionKey)
if agent == nil {
return "", fmt.Errorf("no agent available for session %q", sessionKey)
return agentResponse{}, fmt.Errorf("no agent available for session %q", sessionKey)
}
if tool, ok := agent.Tools.Get("message"); ok {
@ -352,7 +363,7 @@ func (al *AgentLoop) Continue(ctx context.Context, sessionKey, channel, chatID s
}
}
return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, steeringMsgs)
return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, prevContent, steeringMsgs)
}
func (al *AgentLoop) InterruptGraceful(hint string) error {

184
pkg/agent/steering_test.go
View file

@ -298,12 +298,12 @@ func TestAgentLoop_Continue_NoMessages(t *testing.T) {
t.Fatal("expected provider to be initialized")
}
resp, err := al.Continue(context.Background(), "test-session", "test", "chat1")
resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1", "")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resp != "" {
t.Fatalf("expected empty response for no steering messages, got %q", resp)
if resp.Content != "" {
t.Fatalf("expected empty response for no steering messages, got %q", resp.Content)
}
}
@ -331,12 +331,12 @@ func TestAgentLoop_Continue_WithMessages(t *testing.T) {
al.Steer(providers.Message{Role: "user", Content: "new direction"})
resp, err := al.Continue(context.Background(), "test-session", "test", "chat1")
resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1", "")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if resp != "continued response" {
t.Fatalf("expected 'continued response', got %q", resp)
if resp.Content != "continued response" {
t.Fatalf("expected 'continued response', got %q", resp.Content)
}
}
@ -1073,12 +1073,12 @@ func TestAgentLoop_Continue_PreservesSteeringMedia(t *testing.T) {
t.Fatalf("Steer failed: %v", err)
}
resp, err := al.Continue(context.Background(), sessionKey, "test", "chat1")
resp, err := al.continueResponse(context.Background(), sessionKey, "test", "chat1", "")
if err != nil {
t.Fatalf("Continue failed: %v", err)
t.Fatalf("continueResponse failed: %v", err)
}
if resp != "ack" {
t.Fatalf("expected ack, got %q", resp)
if resp.Content != "ack" {
t.Fatalf("expected ack, got %q", resp.Content)
}
capMu.Lock()
@ -1583,6 +1583,170 @@ func (w *wrappingProvider) GetDefaultModel() string {
return w.inner.GetDefaultModel()
}
// TestContinueResponse_EphemeralPrefixBeforePersistence verifies the two
// invariants of the ephemeral-prefix design:
//
// 1. The steering continuation's LLM call sees the previous assistant reply
// in its message list even though OnDelivered has not fired yet (i.e. the
// reply is not yet in session history).
//
// 2. Session history does not contain the assistant reply at the time the
// continuation starts — only ephemeral context was injected.
func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) {
tmpDir, err := os.MkdirTemp("", "agent-test-*")
if err != nil {
t.Fatalf("MkdirTemp: %v", err)
}
defer os.RemoveAll(tmpDir)
cfg := &config.Config{
Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{
Workspace: tmpDir,
ModelName: "test-model",
MaxTokens: 4096,
MaxToolIterations: 10,
},
},
}
msgBus := bus.NewMessageBus()
// Two-call provider: first returns the initial reply, second (for the
// steering continuation) records its message list so we can inspect it.
const firstReply = "first assistant reply"
const continuationReply = "continuation reply"
var capturedMessages []providers.Message
var captureMu sync.Mutex
// sequencedProvider returns firstReply on call 1, continuationReply on call 2,
// and records the message list for the second (steering continuation) call.
seqProv := &sequencedProvider{
responses: []string{firstReply, continuationReply},
onChat: func(n int, msgs []providers.Message) {
if n == 2 {
captureMu.Lock()
capturedMessages = append([]providers.Message(nil), msgs...)
captureMu.Unlock()
}
},
}
al := NewAgentLoop(cfg, msgBus, seqProv)
ctx := context.Background()
const sessionKey = "agent:main:main"
const channel = "test"
const chatID = "chat1"
// Step 1: process the initial user message.
// Do NOT call OnDelivered — the assistant reply must remain unpersisted.
msg := bus.InboundMessage{
Channel: channel,
SenderID: "user1",
ChatID: chatID,
Content: "hello",
}
response, err := al.processMessage(ctx, msg)
if err != nil {
t.Fatalf("processMessage: %v", err)
}
if response.Content != firstReply {
t.Fatalf("first reply = %q, want %q", response.Content, firstReply)
}
// Confirm: assistant message is NOT in session history yet.
agent := al.registry.GetDefaultAgent()
histBefore := agent.Sessions.GetHistory(sessionKey)
for _, m := range histBefore {
if m.Role == "assistant" {
t.Fatalf("assistant in history before OnDelivered: %+v", m)
}
}
// Step 2: enqueue a steering message and run the continuation, passing
// the previous assistant reply as ephemeral context.
if pushErr := al.steering.pushScope(sessionKey, providers.Message{
Role: "user",
Content: "follow-up question",
}); pushErr != nil {
t.Fatalf("pushScope: %v", pushErr)
}
continued, err := al.continueResponse(ctx, sessionKey, channel, chatID, firstReply)
if err != nil {
t.Fatalf("continueResponse: %v", err)
}
if continued.Content != continuationReply {
t.Fatalf("continuation reply = %q, want %q", continued.Content, continuationReply)
}
// Step 3: verify the LLM received the ephemeral assistant reply in context.
captureMu.Lock()
msgs := append([]providers.Message(nil), capturedMessages...)
captureMu.Unlock()
if len(msgs) == 0 {
t.Fatal("no messages captured for the continuation LLM call")
}
var foundEphemeral bool
for _, m := range msgs {
if m.Role == "assistant" && m.Content == firstReply {
foundEphemeral = true
break
}
}
if !foundEphemeral {
roles := make([]string, len(msgs))
for i, m := range msgs {
roles[i] = fmt.Sprintf("%s:%q", m.Role, m.Content)
}
t.Fatalf("ephemeral assistant reply not found in LLM context; messages: %v", roles)
}
// Step 4: verify the session history still does not contain the first
// assistant reply (OnDelivered was never called for it).
histAfter := agent.Sessions.GetHistory(sessionKey)
for _, m := range histAfter {
if m.Role == "assistant" && m.Content == firstReply {
t.Fatal("first assistant reply persisted to history before OnDelivered")
}
}
}
// sequencedProvider returns responses in order, calling onChat(n, msgs) where
// n is the 1-based call index.
type sequencedProvider struct {
responses []string
onChat func(n int, msgs []providers.Message)
mu sync.Mutex
calls int
}
func (s *sequencedProvider) Chat(
ctx context.Context,
messages []providers.Message,
tools []providers.ToolDefinition,
model string,
opts map[string]any,
) (*providers.LLMResponse, error) {
s.mu.Lock()
s.calls++
n := s.calls
resp := ""
if n <= len(s.responses) {
resp = s.responses[n-1]
}
s.mu.Unlock()
if s.onChat != nil {
s.onChat(n, messages)
}
return &providers.LLMResponse{Content: resp, ToolCalls: []providers.ToolCall{}}, nil
}
func (s *sequencedProvider) GetDefaultModel() string { return "seq-model" }
// Ensure NormalizeToolCall handles our test tool calls.
func init() {
// This is a no-op init; we just need the tool call tests to work

7
pkg/bus/types.go
View file

@ -12,7 +12,9 @@ type SenderInfo struct {
PlatformID string `json:"platform_id,omitempty"` // raw platform ID, e.g. "123456"
CanonicalID string `json:"canonical_id,omitempty"` // "platform:id" format
Username string `json:"username,omitempty"` // username (e.g. @alice)
DisplayName string `json:"display_name,omitempty"` // display name
DisplayName string `json:"display_name,omitempty"` // display name (used when first/last are not available)
FirstName string `json:"first_name,omitempty"` // given name (preferred over DisplayName when set)
LastName string `json:"last_name,omitempty"` // family name
}
type InboundMessage struct {
@ -27,6 +29,8 @@ type InboundMessage struct {
MediaScope string `json:"media_scope,omitempty"` // media lifecycle scope
SessionKey string `json:"session_key"`
Metadata map[string]string `json:"metadata,omitempty"`
ReplyToMessageID string `json:"reply_to_message_id,omitempty"` // parent platform message ID
}
type OutboundMessage struct {
@ -35,6 +39,7 @@ type OutboundMessage struct {
Content string `json:"content"`
ReplyToMessageID string `json:"reply_to_message_id,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
OnDelivered func(msgIDs []string) `json:"-"`
}
// MediaPart describes a single media attachment to send.

18
pkg/channels/manager.go
View file

@ -713,11 +713,25 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
chunks = splitByLength(msg.Content, maxLen)
}
// Step 3: Send all chunks
// Step 3: Send all chunks, collecting delivered platform message IDs.
// OnDelivered fires only when every chunk is delivered successfully:
// a partial delivery means the user saw a truncated message, so we
// must not record the full assistant reply as persisted.
var deliveredIDs []string
allDelivered := true
for _, chunk := range chunks {
chunkMsg := msg
chunkMsg.Content = chunk
m.sendWithRetry(ctx, name, w, chunkMsg)
chunkMsg.OnDelivered = nil // delivery callback fires once after all chunks
ids, ok := m.sendWithRetry(ctx, name, w, chunkMsg)
if !ok {
allDelivered = false
break
}
deliveredIDs = append(deliveredIDs, ids...)
}
if allDelivered && msg.OnDelivered != nil {
msg.OnDelivered(deliveredIDs)
}
case <-ctx.Done():
return

2
pkg/channels/telegram/telegram.go
View file

@ -590,6 +590,8 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes
CanonicalID: identity.BuildCanonicalID("telegram", platformID),
Username: user.Username,
DisplayName: user.FirstName,
FirstName: user.FirstName,
LastName: user.LastName,
}
// check allowlist to avoid downloading attachments for rejected users

12
pkg/providers/protocoltypes/types.go
View file

@ -62,6 +62,15 @@ type ContentBlock struct {
CacheControl *CacheControl `json:"cache_control,omitempty"`
}
// MessageSender carries author identity for a user message.
// Stored alongside the message in history so the LLM can address
// participants by name in multi-user conversations.
type MessageSender struct {
Username string `json:"username,omitempty"` // e.g. "@alice" (platform handle)
FirstName string `json:"first_name,omitempty"` // given name
LastName string `json:"last_name,omitempty"` // family name
}
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
@ -70,6 +79,9 @@ type Message struct {
SystemParts []ContentBlock `json:"system_parts,omitempty"` // structured system blocks for cache-aware adapters
ToolCalls []ToolCall `json:"tool_calls,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
MessageIDs []string `json:"message_ids,omitempty"` // Platform message IDs
ReplyToMessageID string `json:"reply_to_message_id,omitempty"` // Parent message ID (for threading)
Sender *MessageSender `json:"sender,omitempty"` // Author identity (user messages only)
}
type ToolDefinition struct {

1
pkg/providers/types.go
View file

@ -19,6 +19,7 @@ type (
GoogleExtra = protocoltypes.GoogleExtra
ContentBlock = protocoltypes.ContentBlock
CacheControl = protocoltypes.CacheControl
MessageSender = protocoltypes.MessageSender
)
type LLMProvider interface {