diff --git a/services/agent-runner/internal/acpbridge/dispatch.go b/services/agent-runner/internal/acpbridge/dispatch.go index 196f45ded..b11a8ea94 100644 --- a/services/agent-runner/internal/acpbridge/dispatch.go +++ b/services/agent-runner/internal/acpbridge/dispatch.go @@ -93,6 +93,7 @@ func (d *Dispatcher) failOffline(ctx context.Context, trigger agent.Trigger) err if err := d.ConvRepo.UpdateStatus(ctx, trigger.ConversationID, "failed", &errMsg); err != nil { return err } + d.publishQueueStatus(ctx, trigger.ConversationID, "failed") if err := d.Publisher.PublishRealtime(ctx, trigger.ProjectID, trigger.ConversationID, "agent.conversation.failed", nil, trigger.ActorUserID); err != nil { d.Log.Warn("acpbridge: failed to publish realtime status", @@ -128,6 +129,7 @@ func (d *Dispatcher) watchdog(conversationID, projectID uuid.UUID, timeout time. } d.Log.Warn("acpbridge: ACP turn timed out with no turn_status from the bridge", "conversation_id", conversationID, "timeout", timeout) + d.publishQueueStatus(ctx, conversationID, "failed") if err := d.Publisher.PublishRealtime(ctx, projectID, conversationID, "agent.conversation.failed", nil, actorUserID); err != nil { d.Log.Warn("acpbridge: failed to publish realtime status", @@ -135,6 +137,28 @@ func (d *Dispatcher) watchdog(conversationID, projectID uuid.UUID, timeout time. } } +// publishQueueStatus tells services/api's AgentQueueConsumer that an ACP +// conversation reached a terminal status, so AdvanceQueue can start whatever +// is queued behind this agent. handler.publishTerminalStatus does the same for +// sandboxed agents; without it here, an ACP agent's backlog is only ever +// drained by a StopConversation. Best-effort, like every publish in this +// file: the DB status the caller just wrote stays the source of truth. +// AdvanceQueue tolerates a duplicate event (see claimQueuedForDispatch), so a +// late turn_status racing the watchdog is harmless. +func (d *Dispatcher) publishQueueStatus(ctx context.Context, conversationID uuid.UUID, status string) { + if err := d.Publisher.PublishConversationStatus(ctx, conversationID, status); err != nil { + d.Log.Warn("acpbridge: failed to publish conversation status", + "conversation_id", conversationID, "status", status, "error", err) + } +} + +// isTerminalStatus mirrors services/api's agentdom.ConversationStatus.IsTerminal: +// the statuses a conversation never leaves, and the only ones AdvanceQueue +// should hear about ("paused" and "running" are not). +func isTerminalStatus(status string) bool { + return status == "finished" || status == "failed" || status == "stopped" +} + func projectIDOrEmpty(id uuid.UUID) string { if id == uuid.Nil { return "" diff --git a/services/agent-runner/internal/acpbridge/queue_status_test.go b/services/agent-runner/internal/acpbridge/queue_status_test.go new file mode 100644 index 000000000..2135ecfbd --- /dev/null +++ b/services/agent-runner/internal/acpbridge/queue_status_test.go @@ -0,0 +1,82 @@ +package acpbridge + +import ( + "context" + "io" + "log/slog" + "testing" + + "github.com/alicebob/miniredis/v2" + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + + "github.com/Paca-AI/agent-runner/internal/messaging" +) + +// An ACP conversation's terminal status has to land on +// StreamAgentConversationStatus — services/api's AgentQueueConsumer runs +// AdvanceQueue only on that stream, so without the entry a conversation queued +// behind the same agent is never started. +func TestPublishQueueStatus_AppendsToConversationStatusStream(t *testing.T) { + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("miniredis.Run: %v", err) + } + t.Cleanup(mr.Close) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { client.Close() }) + d := &Dispatcher{ + Publisher: messaging.NewPublisher(client), + Log: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + + convID := uuid.New() + d.publishQueueStatus(context.Background(), convID, "failed") + + entries, err := client.XRange(context.Background(), messaging.StreamAgentConversationStatus, "-", "+").Result() + if err != nil { + t.Fatalf("XRange: %v", err) + } + if len(entries) != 1 { + t.Fatalf("stream has %d entries, want 1", len(entries)) + } + if got := entries[0].Values["conversation_id"]; got != convID.String() { + t.Errorf("conversation_id = %v, want %s", got, convID) + } + if got := entries[0].Values["status"]; got != "failed" { + t.Errorf("status = %v, want failed", got) + } +} + +// A publish failure is logged, never panics or blocks the caller: the DB +// status is already written and stays the source of truth. +func TestPublishQueueStatus_ToleratesUnreachableValkey(t *testing.T) { + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("miniredis.Run: %v", err) + } + client := redis.NewClient(&redis.Options{Addr: mr.Addr(), MaxRetries: -1}) + t.Cleanup(func() { client.Close() }) + mr.Close() + d := &Dispatcher{ + Publisher: messaging.NewPublisher(client), + Log: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + d.publishQueueStatus(context.Background(), uuid.New(), "finished") +} + +func TestIsTerminalStatus(t *testing.T) { + for status, want := range map[string]bool{ + "finished": true, + "failed": true, + "stopped": true, + "paused": false, + "running": false, + "queued": false, + "": false, + } { + if got := isTerminalStatus(status); got != want { + t.Errorf("isTerminalStatus(%q) = %v, want %v", status, got, want) + } + } +} diff --git a/services/agent-runner/internal/acpbridge/server.go b/services/agent-runner/internal/acpbridge/server.go index 922407046..6ef6626fb 100644 --- a/services/agent-runner/internal/acpbridge/server.go +++ b/services/agent-runner/internal/acpbridge/server.go @@ -353,6 +353,18 @@ func (s *Server) handleTurnStatusMessage(ctx context.Context, agentID uuid.UUID, s.Log.Warn("acpbridge: failed to record turn_status", "conversation_id", convID, "error", err) return } + // A terminal turn_status must also reach StreamAgentConversationStatus, the + // same as handler.publishTerminalStatus does for sandboxed agents: it is the + // only event services/api's AgentQueueConsumer runs AdvanceQueue on. Without + // it, a conversation queued behind this agent stays "queued" forever once the + // running turn ends — only a StopConversation (which publishes on its own) + // would ever free it. Published before the realtime lookup below so a failure + // there can't swallow it. + if isTerminalStatus(statusStr) { + if err := s.Publisher.PublishConversationStatus(ctx, convID, statusStr); err != nil { + s.Log.Warn("acpbridge: failed to publish conversation status", "conversation_id", convID, "status", statusStr, "error", err) + } + } projectID, ownerUserID, err := s.ConvRepo.GetConversationRealtimeContext(ctx, convID) if err != nil {