fix(orchestrator): acknowledge queued messages delivered to passthrough sessions (#3912)

Signed-off-by: Nicolas <nl@blackfuel.ai>
2026-09-24 22:19:24 +02:00
parent 6676d46175
commit cb591c303c
5 changed files with 180 additions and 2 deletions
@@ -1122,8 +1122,9 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 				releaseReservation()
 				return
 			}
-			return
 		}
+		// A delivered (or empty) reservation is acknowledged; a reservation
+		// left in place is re-reserved and re-delivered at every turn end.
 		if err := s.messageQueue.AcknowledgeQueuedForSession(
 			context.WithoutCancel(ctx), identity, queuedMsg,
 		); err != nil &&
@@ -126,6 +126,33 @@ func TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent(t *tes
 	require.Equal(t, models.TaskSessionStateRunning, updatedSession.State)
 }

+func TestHandleAgentReady_PassthroughQueuedMessageDeliveredOnce(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	agentManager := &mockAgentManager{isPassthrough: true}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+	svc.messageQueue.SetAutoMergeEnabled(false)
+
+	first, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "peer report", "", "test", false, nil)
+	require.NoError(t, err)
+	second, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "next report", "", "test", false, nil)
+	require.NoError(t, err)
+
+	for range 3 {
+		svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
+	}
+
+	require.Len(t, agentManager.passthroughStdinCalls, 2, "each queued message must be delivered exactly once")
+	require.Equal(t, "peer report\r", agentManager.passthroughStdinCalls[0].Data)
+	require.Equal(t, "next report\r", agentManager.passthroughStdinCalls[1].Data)
+	for _, id := range []string{first.ID, second.ID} {
+		_, err = svc.messageQueue.FindEntryByID(ctx, id)
+		require.ErrorIs(t, err, messagequeue.ErrEntryNotFound)
+	}
+}
+
 func TestHandleAgentReady_PassthroughQueuedMessagePublishesAfterWriteFailure(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)
@@ -0,0 +1,62 @@
+---
+created: 2026-09-24
+status: done
+requirements:
+  - REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001
+system_design:
+  - ../../specs/tasks/system-design/passthrough-queued-prompt-dispatch.md
+legacy_specs: []
+---
+
+# Implementation Plan: Passthrough Queued Prompt Acknowledgement
+
+## Overview
+
+The ordinary passthrough ready drain reserves the queue head with a retained
+reservation. After a successful PTY write it returned before acknowledging the
+reservation. Every later turn end therefore re-reserved the same entry and
+wrote it again, and the entries behind it were never delivered. One work order
+restores the acknowledgement and adds the regression proof.
+
+## Scope
+
+### In scope
+
+- Acknowledge the reservation after a successful PTY write of a text-only
+  queued prompt in `handleAgentReady`.
+- Add a regression test over repeated `agent.ready` events.
+
+### Out of scope
+
+- A user-facing action to purge a reserved queue entry.
+- Durable lifecycle and attachment-bearing entries, which already acknowledge
+  through the guarded executor path.
+- Changes to the write-failure path, which already releases or requeues.
+
+## Technical approach
+
+Remove the early `return` after a successful write in the ordinary passthrough
+branch of `handleAgentReady`. The success path then falls through to the
+existing `AcknowledgeQueuedForSession` call, which already served the
+empty-prompt case, with the same error filtering.
+
+## Tests
+
+| Acceptance criterion | Evidence |
+| --- | --- |
+| `AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.1` | `TestHandleAgentReady_PassthroughQueuedMessageDeliveredOnce` queues two prompts and sends three `agent.ready` events. It asserts one PTY write per prompt, in order, and that neither entry remains in storage, reserved or otherwise. |
+
+## Work orders
+
+- [x] [Task 01: Acknowledge delivered passthrough queue reservations](task-01-acknowledge-delivered-reservation.md)
+
+## Verification results
+
+- `go test ./internal/orchestrator -run TestHandleAgentReady_PassthroughQueuedMessageDeliveredOnce -count=1` fails before the fix (the first prompt is written three times and the second never) and passes after it.
+- `go test ./internal/orchestrator/... -count=1` passed.
+
+## Risks
+
+- A durable plan comment is now acknowledged right after a successful write.
+  This matches the attempted-delivery acknowledgement already made by the
+  write-failure path.
@@ -0,0 +1,84 @@
+---
+id: "01-acknowledge-delivered-reservation"
+title: "Acknowledge delivered passthrough queue reservations"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001
+acceptance_criteria:
+  - AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.1
+system_design:
+  - ../../specs/tasks/system-design/passthrough-queued-prompt-dispatch.md
+---
+
+# Task 01: Acknowledge Delivered Passthrough Queue Reservations
+
+## Summary
+
+Deliver a queued prompt to a passthrough session exactly once. After a
+successful PTY write, acknowledge the retained reservation so the next turn end
+selects the next entry instead of rewriting the same one.
+
+## In scope
+
+- The ordinary text-only branch of the passthrough ready drain in
+  `handleAgentReady`.
+- A regression test with repeated `agent.ready` events.
+
+## Out of scope
+
+- Purge actions for reserved entries.
+- Durable lifecycle and attachment-bearing queue delivery.
+
+## Acceptance
+
+- Three consecutive `agent.ready` events after two queued prompts produce one
+  PTY write per prompt, in queue order.
+- Each delivered entry is removed from storage, not only hidden from the visible
+  queue count.
+- A failed PTY write still releases the reservation, or requeues it for plan
+  comments.
+
+## Verification
+
+```bash
+go test ./internal/orchestrator -run 'TestHandleAgentReady_Passthrough' -count=1
+go test ./internal/orchestrator/... -count=1
+```
+
+Run both commands from `apps/backend`.
+
+## Files likely touched
+
+- `apps/backend/internal/orchestrator/event_handlers_agent.go`
+- `apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- Acknowledging before the CLI consumes the paste would lose a prompt if the
+  CLI discards it. The PTY write is the delivery boundary for passthrough
+  sessions, as it is for the empty-prompt acknowledgement.
+
+## Parallelism
+
+`sequential`
+
+## Inputs
+
+- `docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md`
+- `docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md`
+
+## Results
+
+- Removed the early return after a successful write, so the existing
+  acknowledgement runs for delivered prompts.
+- Added `TestHandleAgentReady_PassthroughQueuedMessageDeliveredOnce`. It fails
+  before the fix, where the first prompt is written three times and the second
+  is starved, and passes after it.
+- `go test ./internal/orchestrator/... -count=1` passed.
@@ -67,7 +67,11 @@ immediate marker contract.
   publishing it.
 3. While the guard is still held, the orchestrator writes all planned prompt
   chunks to the PTY. This preserves the existing exclusion against a
-   concurrent cancel, manual drain, or prompt dispatch.
+   concurrent cancel, manual drain, or prompt dispatch. After a successful
+   write, the orchestrator acknowledges the reservation. The reservation is
+   retained in durable storage until acknowledged, so an unacknowledged entry
+   is re-reserved and written again at every later turn end, and it blocks the
+   entries queued behind it.
 4. The ready handler releases the session guard.
 5. The ready handler invokes the deferred publication callback synchronously.
   `MemoryEventBus` calls `handleAgentRunning`, which can now acquire the free