fix(agent-runner): drain the parallelism queue when an ACP conversation ends

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,13 +129,36 @@ 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",
 			"conversation_id", conversationID, "error", err)
 	}
 }
 
+// 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 ""

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

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 {