From cb591c303c437946854c1d509cee4a3228f9da86 Mon Sep 17 00:00:00 2001 From: Nicolas Lenepveu Date: Thu, 24 Sep 2026 22:19:24 +0200 Subject: [PATCH] fix(orchestrator): acknowledge queued messages delivered to passthrough sessions (#3912) Signed-off-by: Nicolas --- .../orchestrator/event_handlers_agent.go | 3 +- ...event_handlers_passthrough_running_test.go | 27 ++++++ .../passthrough-queued-prompt-ack/plan.md | 62 ++++++++++++++ ...sk-01-acknowledge-delivered-reservation.md | 84 +++++++++++++++++++ .../passthrough-queued-prompt-dispatch.md | 6 +- 5 files changed, 180 insertions(+), 2 deletions(-) create mode 100644 docs/plans/passthrough-queued-prompt-ack/plan.md create mode 100644 docs/plans/passthrough-queued-prompt-ack/task-01-acknowledge-delivered-reservation.md diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 6f7f511a87..db507811c0 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -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 && diff --git a/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go b/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go index 2b6ea8e39e..2531571f9b 100644 --- a/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go @@ -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) diff --git a/docs/plans/passthrough-queued-prompt-ack/plan.md b/docs/plans/passthrough-queued-prompt-ack/plan.md new file mode 100644 index 0000000000..803da861a4 --- /dev/null +++ b/docs/plans/passthrough-queued-prompt-ack/plan.md @@ -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. diff --git a/docs/plans/passthrough-queued-prompt-ack/task-01-acknowledge-delivered-reservation.md b/docs/plans/passthrough-queued-prompt-ack/task-01-acknowledge-delivered-reservation.md new file mode 100644 index 0000000000..3bc19f4afc --- /dev/null +++ b/docs/plans/passthrough-queued-prompt-ack/task-01-acknowledge-delivered-reservation.md @@ -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. diff --git a/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md b/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md index dfb028c51c..ad34b7c5bf 100644 --- a/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md +++ b/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md @@ -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