diff --git a/apps/backend/internal/agent/runtime/lifecycle/session.go b/apps/backend/internal/agent/runtime/lifecycle/session.go index 0f096385b3..fb89a7f757 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session.go @@ -32,6 +32,16 @@ const ( pendingDispatchedPromptWaitTimeout = 10 * time.Second ) +// ErrSteerNotDispatched reports that a steer found no active prompt generation +// to reuse. The orchestrator must run the ordinary prompt admission path when +// it receives this result because a new turn may have become promptable. +var ErrSteerNotDispatched = errors.New("steer was not dispatched") + +// ErrSteerAttachmentMaterialization wraps an attachment delivery failure that +// happened before agentctl accepted the steer. The orchestrator uses this +// identity to release the steer slot and retry a queued successor drain. +var ErrSteerAttachmentMaterialization = errors.New("steer attachment materialization failed") + // PendingDispatchedPromptTimeoutError reports that a successor prompt could // not observe completion of an earlier dispatch-only prompt within the // bounded admission interval. The pending gate remains set so cancellation @@ -1243,8 +1253,9 @@ func (sm *SessionManager) markPromptDispatched(execution *AgentExecution, genera // // Delivery is opportunistic: with no live turn to fold into (idle, a turn that // was admitted but not yet dispatched, one that already completed, or a -// disconnected stream) there is nothing to steer, so it falls back to an -// ordinary prompt delivered in submission order rather than dropping the message. +// disconnected stream) there is nothing to steer. The orchestrator receives +// ErrSteerNotDispatched and re-enters ordinary prompt admission in submission +// order rather than sending around the queue. func (sm *SessionManager) SendPromptSteerWithDispatchCallback( ctx context.Context, execution *AgentExecution, @@ -1263,21 +1274,32 @@ func (sm *SessionManager) SendPromptSteerWithDispatchCallback( return nil, fmt.Errorf("execution %q is not ready for prompts (status: %s)", execution.ID, execution.Status) } - steered, err := sm.tryDispatchSteer(ctx, execution, prompt, attachments) + // Avoid an upload when there is no generation to steer into. This is only an + // optimization; the lock-guarded dispatch below remains authoritative for + // the race where the active generation completes during materialization. + if sm.activePromptGeneration(execution) == 0 { + return nil, ErrSteerNotDispatched + } + + // Resolve claimed descriptors before the steer takes promptLifecycleMu. + // Materialization streams file bytes to agentctl and dispatchSteerLocked holds + // that lock under a bounded timeout, so uploading inside it would let a large + // attachment block the lifecycle lock for the length of a network write. + // + // A failure here returns instead of falling back: the ordinary prompt route + // would carry the same unresolved descriptor, and the agent would receive a + // reference to bytes that are absent from the session. + materialized, err := sm.materializeAttachments(ctx, execution, attachments) + if err != nil { + return nil, fmt.Errorf("%w: %w", ErrSteerAttachmentMaterialization, err) + } + + steered, err := sm.tryDispatchSteer(ctx, execution, prompt, materialized) if err != nil { return nil, err } if !steered { - return sm.sendPrompt( - ctx, - execution, - prompt, - validateStatus, - attachments, - dispatchOnly, - sendPromptCallbacks{onDispatched: onDispatched}, - false, - ) + return nil, ErrSteerNotDispatched } if onDispatched != nil { onDispatched() @@ -1293,8 +1315,8 @@ func (sm *SessionManager) SendPromptSteerWithDispatchCallback( // round-trip (agentctl acks before it runs the prompt), but sendStreamRequest has // no deadline of its own, so a half-open connection could otherwise pin the lock // — wedging this execution's completion claims and new prompts — until the stream -// tears down. A timeout lands on the error path, which is send-safe under the -// no-resend fallback. Exposed as a var so tests can shorten it. +// tears down. A timeout lands on the error path without retrying the steer. +// Exposed as a var so tests can shorten it. var steerDispatchTimeout = 5 * time.Second // tryDispatchSteer selects the active dispatched generation and issues the steer @@ -1313,7 +1335,8 @@ var steerDispatchTimeout = 5 * time.Second // stalled connection, and the history append is done after the lock is released. // // Returns (false, nil) when there is nothing to steer into — no live generation, -// or the stream is not connected — so the caller degrades to an ordinary prompt. +// or the stream is not connected — so the orchestrator can admit an ordinary +// prompt. // It deliberately uses the single fast RPC, not dispatchPrompt's reconnect path: // a disconnected stream means there is no live turn to fold into, and reconnect // can block for seconds, which must not run under the lifecycle lock. @@ -1371,9 +1394,8 @@ func (sm *SessionManager) dispatchSteerLocked( } // recordSteerActivity mirrors the ordinary prompt path's history append and -// activity-timestamp bump for a steer that actually dispatched. Kept off the -// fallback path so a steer that degrades to an ordinary prompt is not recorded -// twice. +// activity-timestamp bump for a steer that actually dispatched. A no-dispatch +// result never reaches this method, so ordinary fallback is not recorded twice. // // Deliberately does not touch agentEventSincePrompt: a steer is user input // injected into an already-running turn, not agent output, so it must not diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go index b8556d7ec7..cf42724ef9 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go @@ -2,12 +2,16 @@ package lifecycle import ( "context" + "errors" + "io" + "strings" "sync" "testing" "time" agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" "github.com/kandev/kandev/internal/agentctl/types/streams" + v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) @@ -15,8 +19,8 @@ import ( // (SessionManager.SendPromptSteerWithDispatchCallback). The invariant under test: // a steer reuses the *active dispatched* prompt generation and never serializes // behind it — so the operator's input reaches the still-generating turn — while -// a steer with no live turn (idle, pre-dispatch, or already-completed) degrades -// to an ordinary prompt delivered in submission order. +// a steer with no live turn (idle, pre-dispatch, or already-completed) returns a +// no-dispatch result for ordinary admission to handle in submission order. type capturedPrompt struct { text string @@ -24,6 +28,31 @@ type capturedPrompt struct { steer bool } +type blockingAttachmentReader struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func (r *blockingAttachmentReader) OpenClaimed(context.Context, string, string, string) (io.ReadCloser, string, string, int64, error) { + return &blockingAttachmentReadCloser{reader: r, data: strings.NewReader("attachment bytes")}, "bundle.zip", "application/zip", 16, nil +} + +type blockingAttachmentReadCloser struct { + reader *blockingAttachmentReader + data *strings.Reader +} + +func (r *blockingAttachmentReadCloser) Read(p []byte) (int, error) { + r.reader.once.Do(func() { close(r.reader.started) }) + <-r.reader.release + return r.data.Read(p) +} + +func (r *blockingAttachmentReadCloser) Close() error { + return nil +} + // promptProbe captures every agent.prompt frame the mock receives so a test can // assert the wire-level generation and steer flag. type promptProbe struct { @@ -83,6 +112,7 @@ type steerHarness struct { mgr *Manager exec *AgentExecution probe *promptProbe + mock *mockAgentServer ctx context.Context } @@ -106,10 +136,15 @@ func newSteerHarness(t *testing.T) *steerHarness { exec := createTestExecution("exec-steer", "task-steer", "session-steer") exec.agentctl = client + // Attachment materialization is addressed to the ACP session and refuses to + // run without one, so every steer harness carries a session identity and a + // reader even when the test under it sends no attachments. + exec.ACPSessionID = "acp-session-steer" + mgr.sessionManager.SetAttachmentReader(testAttachmentReader{}) if err := mgr.executionStore.Add(exec); err != nil { t.Fatalf("add execution: %v", err) } - return &steerHarness{mgr: mgr, exec: exec, probe: probe, ctx: ctx} + return &steerHarness{mgr: mgr, exec: exec, probe: probe, mock: mock, ctx: ctx} } // startForegroundTurn dispatches an ordinary prompt and blocks it in flight @@ -206,9 +241,9 @@ func TestSendPromptSteer_ReusesActiveGenerationWithoutBlocking(t *testing.T) { } } -// TestSendPromptSteer_FallsBackWhenNoTurnInFlight proves an idle steer-capable -// session degrades to an ordinary prompt (a new generation, steer=false). -func TestSendPromptSteer_FallsBackWhenNoTurnInFlight(t *testing.T) { +// TestSendPromptSteer_ReturnsNotDispatchedWhenNoTurnInFlight proves an idle +// steer-capable session leaves ordinary prompt admission to the orchestrator. +func TestSendPromptSteer_ReturnsNotDispatchedWhenNoTurnInFlight(t *testing.T) { h := newSteerHarness(t) if got := h.mgr.executionStore.ActivePromptGeneration(h.exec.ID); got != 0 { @@ -218,25 +253,20 @@ func TestSendPromptSteer_FallsBackWhenNoTurnInFlight(t *testing.T) { res, err := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( h.ctx, h.exec, "no turn yet", false, nil, true, nil, ) - if err != nil { - t.Fatalf("steer fallback errored: %v", err) + if !errors.Is(err, ErrSteerNotDispatched) { + t.Fatalf("steer result = %+v, error = %v, want ErrSteerNotDispatched", res, err) } - if res == nil || res.StopReason != PromptStopReasonDispatched { - t.Fatalf("steer fallback result = %+v, want dispatched", res) - } - sp := h.probe.next(t) - if sp.steer { - t.Fatal("fallback prompt must not carry steer=true") - } - if sp.generation != 1 { - t.Fatalf("fallback generation = %d, want 1 (a fresh ordinary prompt)", sp.generation) + select { + case frame := <-h.probe.ch: + t.Fatalf("idle steer dispatched a prompt outside ordinary admission: %+v", frame) + case <-time.After(200 * time.Millisecond): } } -// TestSendPromptSteer_FallsBackAfterPredecessorCompleted proves a steer that -// arrives after the foreground turn already finished does not attach to that -// dead generation; it runs as a fresh ordinary prompt. -func TestSendPromptSteer_FallsBackAfterPredecessorCompleted(t *testing.T) { +// TestSendPromptSteer_ReturnsNotDispatchedAfterPredecessorCompleted proves a +// steer that arrives after the foreground turn already finished does not attach +// to that dead generation or dispatch around ordinary admission. +func TestSendPromptSteer_ReturnsNotDispatchedAfterPredecessorCompleted(t *testing.T) { h := newSteerHarness(t) firstDone := h.startForegroundTurn(t, 1) @@ -257,33 +287,26 @@ func TestSendPromptSteer_FallsBackAfterPredecessorCompleted(t *testing.T) { res, err := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( h.ctx, h.exec, "after done", false, nil, true, nil, ) - if err != nil { - t.Fatalf("steer-after-completion errored: %v", err) + if !errors.Is(err, ErrSteerNotDispatched) { + t.Fatalf("steer result = %+v, error = %v, want ErrSteerNotDispatched", res, err) } - if res == nil || res.StopReason != PromptStopReasonDispatched { - t.Fatalf("steer-after-completion result = %+v, want dispatched", res) - } - sp := h.probe.next(t) - if sp.steer { - t.Fatal("post-completion steer must fall back to an ordinary prompt (steer=false)") - } - if sp.generation != 2 { - t.Fatalf("post-completion fallback generation = %d, want 2", sp.generation) + select { + case frame := <-h.probe.ch: + t.Fatalf("post-completion steer dispatched around ordinary admission: %+v", frame) + case <-time.After(200 * time.Millisecond): } } // TestSendPromptSteer_DoesNotOvertakeAdmittedButUndispatchedTurn proves the // pre-dispatch race is closed with a real predecessor: while a genuine prompt is // pinned in the admitted-but-undispatched window (generation begun, prompt not -// yet sent), a racing steer must not bind to it or overtake it. The steer falls -// back to an ordinary prompt that serializes behind the predecessor, so the wire -// order is predecessor-then-steer, both ordinary, with no steer frame at all. +// yet sent), a racing steer must not bind to it or overtake it. The lifecycle +// path returns without sending; the orchestrator owns the ordinary fallback. func TestSendPromptSteer_DoesNotOvertakeAdmittedButUndispatchedTurn(t *testing.T) { h := newSteerHarness(t) // Pin the first (predecessor) sendPrompt in the window between BeginPrompt and - // dispatch. Only the first call blocks; the steer's own fallback sendPrompt - // (the second) runs unhindered. + // dispatch. var once sync.Once predecessorPinned := make(chan struct{}) releasePredecessor := make(chan struct{}) @@ -302,15 +325,13 @@ func TestSendPromptSteer_DoesNotOvertakeAdmittedButUndispatchedTurn(t *testing.T <-predecessorPinned // generation 1 admitted, not yet dispatched, promptMu held // The steer arrives now. activePromptGeneration is 0 (gen 1 not marked - // dispatched), so it must fall back — and the fallback then blocks on promptMu - // behind the pinned predecessor. - steerDone := make(chan struct{}) - go func() { - _, _ = h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( - h.ctx, h.exec, "racing steer", false, nil, true, nil, - ) - close(steerDone) - }() + // dispatched), so it returns without sending a frame. + _, steerErr := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( + h.ctx, h.exec, "racing steer", false, nil, true, nil, + ) + if !errors.Is(steerErr, ErrSteerNotDispatched) { + t.Fatalf("steer error = %v, want ErrSteerNotDispatched", steerErr) + } // No frame of any kind can reach the agent while the predecessor is pinned: // the predecessor hasn't dispatched and the steer is blocked behind it. @@ -331,21 +352,17 @@ func TestSendPromptSteer_DoesNotOvertakeAdmittedButUndispatchedTurn(t *testing.T t.Fatalf("first frame = %+v, want ordinary generation 1 (predecessor)", first) } - // The predecessor left a pending dispatched prompt, so the steer's fallback - // waits for generation 1 to complete before it dispatches — driving that - // completion lets it proceed, in order. + // The predecessor left a pending dispatched prompt. No successor was sent by + // this lifecycle call; the orchestrator must decide when to admit it. h.mgr.handleAgentEvent(h.exec, agentctl.AgentEvent{ Type: streams.EventTypeComplete, PromptGeneration: 1, Data: map[string]any{"stop_reason": "end_turn"}, }) - <-steerDone - - // The steer fell back to an ordinary generation 2 — never a steer frame, and - // strictly after the predecessor. - second := h.probe.next(t) - if second.steer || second.generation != 2 { - t.Fatalf("second frame = %+v, want ordinary generation 2 (steer fell back, in order)", second) + select { + case frame := <-h.probe.ch: + t.Fatalf("lifecycle sent a successor frame without ordinary admission: %+v", frame) + case <-time.After(200 * time.Millisecond): } } @@ -467,3 +484,170 @@ func TestSendPromptSteer_NilAgentctlClientErrors(t *testing.T) { t.Fatal("expected an error when the execution has no agentctl client") } } + +// claimedAttachment is the bounded descriptor shape the prompt and queue +// protocols carry. Its bytes live in backend storage until materialization +// streams them to agentctl. +func claimedAttachment() []v1.MessageAttachment { + return []v1.MessageAttachment{ + {AttachmentID: "attachment-steer", Type: "resource", Name: "bundle.zip", SizeBytes: 16}, + } +} + +// TestSendPromptSteer_MaterializesAttachmentsBeforeDispatch is the core +// regression: a steer folded into a still-generating turn must stream its +// claimed bytes to agentctl before the prompt frame goes out. Without that, the +// agent receives a reference to a file that was never written into the session +// and the attachment silently reads as empty. +func TestSendPromptSteer_MaterializesAttachmentsBeforeDispatch(t *testing.T) { + h := newSteerHarness(t) + done := h.startForegroundTurn(t, 1) + + result, err := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( + h.ctx, h.exec, "look at this file", false, claimedAttachment(), false, nil, + ) + if err != nil { + t.Fatalf("steer with attachment: %v", err) + } + if result.StopReason != PromptStopReasonDispatched { + t.Fatalf("stop reason = %q, want %q", result.StopReason, PromptStopReasonDispatched) + } + + steered := h.probe.next(t) + if !steered.steer { + t.Fatalf("frame was not a steer: %+v", steered) + } + + uploads := h.mock.materializedUploads() + if len(uploads) != 1 { + t.Fatalf("materialized uploads = %d, want 1", len(uploads)) + } + if uploads[0].attachmentID != "attachment-steer" { + t.Errorf("attachment_id = %q", uploads[0].attachmentID) + } + if uploads[0].sessionID != "acp-session-steer" { + t.Errorf("session_id = %q", uploads[0].sessionID) + } + if uploads[0].body != "attachment bytes" { + t.Errorf("uploaded body = %q", uploads[0].body) + } + + h.mgr.handleAgentEvent(h.exec, agentctl.AgentEvent{ + Type: streams.EventTypeComplete, + PromptGeneration: 1, + Data: map[string]any{"stop_reason": "end_turn"}, + }) + <-done +} + +// TestSendPromptSteer_DoesNotFallbackOutsideAdmission proves that a turn which +// ends while attachment bytes are materializing does not dispatch a low-level +// ordinary prompt after the steer has lost its generation. The caller must +// return a no-dispatch result so the orchestrator can re-enter normal prompt +// admission and preserve ordering with concurrent messages. +func TestSendPromptSteer_DoesNotFallbackOutsideAdmission(t *testing.T) { + h := newSteerHarness(t) + done := h.startForegroundTurn(t, 1) + reader := &blockingAttachmentReader{started: make(chan struct{}), release: make(chan struct{})} + h.mgr.sessionManager.SetAttachmentReader(reader) + + steerDone := make(chan struct { + result *PromptResult + err error + }, 1) + go func() { + result, err := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( + h.ctx, h.exec, "racing attachment", false, claimedAttachment(), true, nil, + ) + steerDone <- struct { + result *PromptResult + err error + }{result: result, err: err} + }() + <-reader.started + + h.mgr.handleAgentEvent(h.exec, agentctl.AgentEvent{ + Type: streams.EventTypeComplete, + PromptGeneration: 1, + Data: map[string]any{"stop_reason": "end_turn"}, + }) + select { + case err := <-done: + if err != nil { + t.Fatalf("predecessor turn returned error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("predecessor turn did not settle while attachment materialized") + } + + close(reader.release) + select { + case call := <-steerDone: + if !errors.Is(call.err, ErrSteerNotDispatched) { + t.Fatalf("steer result = %+v, error = %v, want ErrSteerNotDispatched", call.result, call.err) + } + case <-time.After(2 * time.Second): + t.Fatal("steer did not return after attachment materialization") + } + select { + case frame := <-h.probe.ch: + t.Fatalf("steer sent a fallback prompt outside orchestrator admission: %+v", frame) + case <-time.After(200 * time.Millisecond): + } +} + +// TestSendPrompt_MaterializesAttachmentsExactlyOnce covers the ordinary route. +// It materializes a claimed descriptor once before dispatching the prompt. +func TestSendPrompt_MaterializesAttachmentsExactlyOnce(t *testing.T) { + h := newSteerHarness(t) + + if _, err := h.mgr.sessionManager.SendPrompt( + h.ctx, h.exec, "ordinary prompt", false, claimedAttachment(), true, + ); err != nil { + t.Fatalf("ordinary prompt: %v", err) + } + + dispatched := h.probe.next(t) + if dispatched.steer { + t.Fatalf("ordinary frame should not be a steer: %+v", dispatched) + } + if uploads := h.mock.materializedUploads(); len(uploads) != 1 { + t.Fatalf("materialized uploads = %d, want exactly 1", len(uploads)) + } +} + +// TestSendPromptSteer_MaterializationFailureDoesNotDispatch pins the failure +// contract. A steer that cannot resolve its bytes must surface an error rather +// than deliver a dangling reference, and must not fall through to an ordinary +// prompt carrying the same unresolved descriptor. +func TestSendPromptSteer_MaterializationFailureDoesNotDispatch(t *testing.T) { + h := newSteerHarness(t) + done := h.startForegroundTurn(t, 1) + h.mock.setFailMaterialize(true) + + _, err := h.mgr.sessionManager.SendPromptSteerWithDispatchCallback( + h.ctx, h.exec, "look at this file", false, claimedAttachment(), false, nil, + ) + if err == nil { + t.Fatal("steer with failing materialization returned no error") + } + if !errors.Is(err, ErrSteerAttachmentMaterialization) { + t.Fatalf("error = %v, want ErrSteerAttachmentMaterialization", err) + } + if !strings.Contains(err.Error(), "materialize attachment") { + t.Fatalf("error = %v, want a materialization error", err) + } + + select { + case frame := <-h.probe.ch: + t.Fatalf("a prompt frame dispatched despite the failure: %+v", frame) + case <-time.After(200 * time.Millisecond): + } + + h.mgr.handleAgentEvent(h.exec, agentctl.AgentEvent{ + Type: streams.EventTypeComplete, + PromptGeneration: 1, + Data: map[string]any{"stop_reason": "end_turn"}, + }) + <-done +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_test.go index aa44fc0dcd..9dfbef68c3 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net" "net/http" "net/http/httptest" @@ -56,6 +57,17 @@ type mockAgentServer struct { upgrader websocket.Upgrader handler func(msg ws.Message) *ws.Message wsConnected chan struct{} // closed when WS stream connects + materialized []materializedUpload + failMaterialize bool +} + +// materializedUpload records one /api/v1/attachments/materialize request so a +// test can assert that the bytes reached agentctl before the prompt dispatched. +type materializedUpload struct { + sessionID string + attachmentID string + name string + body string } func newMockAgentServer(t *testing.T) *mockAgentServer { @@ -154,10 +166,63 @@ func newMockAgentServer(t *testing.T) *mockAgentServer { } }) + // Attachment materialization endpoint. The lifecycle manager streams claimed + // descriptors here before it dispatches a prompt, so recording each upload is + // what lets a steer test prove materialization happened first. + mux.HandleFunc("/api/v1/attachments/materialize", func(w http.ResponseWriter, r *http.Request) { + if err := r.ParseMultipartForm(1 << 20); err != nil { + http.Error(w, "bad multipart", http.StatusBadRequest) + return + } + m.mu.Lock() + fail := m.failMaterialize + m.mu.Unlock() + if fail { + http.Error(w, "materialization unavailable", http.StatusInternalServerError) + return + } + file, _, err := r.FormFile("file") + if err != nil { + http.Error(w, "file is required", http.StatusBadRequest) + return + } + defer func() { _ = file.Close() }() + body, err := io.ReadAll(file) + if err != nil { + http.Error(w, "unreadable file", http.StatusBadRequest) + return + } + name := r.FormValue("name") + m.mu.Lock() + m.materialized = append(m.materialized, materializedUpload{ + sessionID: r.FormValue("session_id"), + attachmentID: r.FormValue("attachment_id"), + name: name, + body: string(body), + }) + m.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"name":%q,"size_bytes":%d}`, name, len(body)) + }) + m.server = httptest.NewServer(mux) return m } +// materializedUploads returns a copy of the recorded materialization requests. +func (m *mockAgentServer) materializedUploads() []materializedUpload { + m.mu.Lock() + defer m.mu.Unlock() + return append([]materializedUpload(nil), m.materialized...) +} + +// setFailMaterialize makes the materialization endpoint reject every upload. +func (m *mockAgentServer) setFailMaterialize(fail bool) { + m.mu.Lock() + m.failMaterialize = fail + m.mu.Unlock() +} + // buildResponse returns the handler or default response for a request message. func (m *mockAgentServer) buildResponse(msg ws.Message) *ws.Message { if m.handler != nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 5b8d582af1..91ddd0d20f 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -320,6 +320,9 @@ type mockAgentManager struct { // steerAgentWithDispatchCallback capability. capturedSteerCalls []promptCall steerErr error + steerStarted chan struct{} + steerRelease chan struct{} + steerStartOnce sync.Once // Optional: closed once on the first PromptAgent call so tests can wait // deterministically without polling. Tests opt in by initializing the channel. promptDone chan struct{} @@ -513,7 +516,15 @@ func (m *mockAgentManager) SteerAgentWithDispatchCallback(_ context.Context, exe m.mu.Lock() m.capturedSteerCalls = append(m.capturedSteerCalls, promptCall{ExecutionID: executionID, Prompt: prompt, DispatchOnly: dispatchOnly}) steerErr := m.steerErr + steerStarted := m.steerStarted + steerRelease := m.steerRelease m.mu.Unlock() + if steerStarted != nil { + m.steerStartOnce.Do(func() { close(steerStarted) }) + } + if steerRelease != nil { + <-steerRelease + } if steerErr != nil { return nil, steerErr } diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction.go b/apps/backend/internal/orchestrator/executor/executor_interaction.go index bf61e75b97..166df63c03 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction.go @@ -286,6 +286,15 @@ const stopReasonPassthrough = "passthrough_dispatched" var ErrPromptDispatchCallbackUnsupported = errors.New("agent manager does not support prompt dispatch callback") +// ErrSteerNotDispatched reports that no active prompt generation accepted a +// steer. The orchestrator must route this outcome through ordinary admission. +var ErrSteerNotDispatched = lifecycle.ErrSteerNotDispatched + +// ErrSteerAttachmentMaterialization identifies a steer attachment failure that +// happened before agentctl accepted the request. The orchestrator uses it to +// retry queued draining after it releases the steer slot. +var ErrSteerAttachmentMaterialization = lifecycle.ErrSteerAttachmentMaterialization + // Prompt sends a follow-up prompt to a running agent for a task // Returns PromptResult indicating if the agent needs input // Attachments (images) are passed to the agent if provided. diff --git a/apps/backend/internal/orchestrator/steer.go b/apps/backend/internal/orchestrator/steer.go index 56b8bc5c74..b540d095eb 100644 --- a/apps/backend/internal/orchestrator/steer.go +++ b/apps/backend/internal/orchestrator/steer.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -74,7 +75,10 @@ func (s *Service) steerEligibleForSession(ctx context.Context, sessionID string) // rather than dispatched now; the caller sees success. ErrSteerNotEligible is // returned only when the session is no longer steer-eligible at all (e.g. its // turn ended between the caller's check and here), so the caller can fall back -// to an ordinary prompt. Any other error is a genuine dispatch failure. +// to an ordinary prompt. If attachment materialization fails after admission, +// the failed owner releases its slot and retries a queued drain so a completion +// that raced with the failure cannot strand the successor. Any other error is a +// genuine dispatch failure. func (s *Service) SteerTask( ctx context.Context, taskID, sessionID, prompt, model string, @@ -142,7 +146,17 @@ func (s *Service) SteerTask( } // We now own the single in-flight slot. The queue admission lock is released // before the blocking dispatch; clear the slot when the dispatch completes. - defer s.steerInFlight.Delete(sessionID) + promptCtx := context.WithoutCancel(ctx) + var dispatchErr error + defer func() { + s.steerInFlight.Delete(sessionID) + if errors.Is(dispatchErr, executor.ErrSteerAttachmentMaterialization) { + // A predecessor completion may have skipped its drain while this + // marker was set. Retry after releasing the marker so the queued + // successor can make progress if the session is promptable now. + s.drainQueuedMessageForPromptableSession(promptCtx, sessionID) + } + }() effectivePrompt := s.effectivePromptForSession(sessionID, prompt, planMode, session) s.logger.Info("dispatching mid-turn steer", @@ -151,19 +165,24 @@ func (s *Service) SteerTask( // context.WithoutCancel: a WS request timeout must not abort the steer, same // as the ordinary prompt path. - promptCtx := context.WithoutCancel(ctx) // dispatchOnly=true: a steer is dispatch-and-continue. The predecessor turn // owns foreground completion, so this call must not wait for a turn to end, // and must not take the foreground claim. - result, err := s.executor.SteerWithDispatchCallback( + result, dispatchErr := s.executor.SteerWithDispatchCallback( promptCtx, taskID, sessionID, effectivePrompt, attachments, true, func() {}, session, ) - if err != nil { + if dispatchErr != nil { + if errors.Is(dispatchErr, executor.ErrSteerNotDispatched) { + // The active generation ended before the steer reached agentctl. + // Let the handler re-enter PromptTask, which owns ordinary queue and + // foreground admission for the replacement turn. + return nil, ErrSteerNotEligible + } s.logger.Warn("mid-turn steer dispatch failed", zap.String("task_id", taskID), zap.String("session_id", sessionID), - zap.Error(err)) - return nil, err + zap.Error(dispatchErr)) + return nil, dispatchErr } return &PromptResult{ StopReason: result.StopReason, diff --git a/apps/backend/internal/orchestrator/steer_test.go b/apps/backend/internal/orchestrator/steer_test.go index b2986780f4..49dc932a12 100644 --- a/apps/backend/internal/orchestrator/steer_test.go +++ b/apps/backend/internal/orchestrator/steer_test.go @@ -9,6 +9,7 @@ import ( "time" "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -333,6 +334,97 @@ func TestSteerTask_DispatchesAndReleasesInFlightSlot(t *testing.T) { t.Fatal("in-flight slot was not released after a dispatch error") } }) + + t.Run("maps a lifecycle no-dispatch result to ordinary admission", func(t *testing.T) { + svc, agentMgr := steerDispatchService(t, taskID, sessionID) + agentMgr.steerErr = executor.ErrSteerNotDispatched + + if _, err := svc.SteerTask(ctx, taskID, sessionID, "steer", "", false, nil); !errors.Is(err, ErrSteerNotEligible) { + t.Fatalf("SteerTask error = %v, want ErrSteerNotEligible", err) + } + if _, inFlight := svc.steerInFlight.Load(sessionID); inFlight { + t.Fatal("lifecycle no-dispatch left an in-flight slot claimed") + } + }) +} + +// TestSteerTask_PreDispatchFailureDrainsSuccessor proves that a queued message +// cannot remain stranded when the admitted steer fails before agentctl accepts +// it. A completion can observe steerInFlight during the materialization or +// dispatch window and skip its drain, so the failed owner must trigger a new +// drain after it releases the slot. +func TestSteerTask_PreDispatchFailureDrainsSuccessor(t *testing.T) { + ctx := context.Background() + const taskID = "task-steer-pre-dispatch-failure" + const sessionID = "session-steer-pre-dispatch-failure" + + steerStarted := make(chan struct{}) + steerRelease := make(chan struct{}) + promptDone := make(chan struct{}) + agentMgr := &mockAgentManager{ + isAgentRunning: true, + steerErr: executor.ErrSteerAttachmentMaterialization, + steerStarted: steerStarted, + steerRelease: steerRelease, + promptDone: promptDone, + } + repo := setupTestRepo(t) + agentMgr.repoForExecutionLookup = repo + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, "exec-steer-pre-dispatch-failure") + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.config.ClaudeMidTurnSteering = true + advertisePromptQueueingForTest(t, svc, sessionID) + svc.registerBackgroundTask(sessionID, "bg-1") + svc.markForegroundIdle(sessionID) + svc.completeBackgroundTask(sessionID, "bg-1") + + firstDone := make(chan error, 1) + go func() { + _, err := svc.SteerTask(ctx, taskID, sessionID, "first steer", "", false, nil) + firstDone <- err + }() + <-steerStarted + + result, err := svc.SteerTask(ctx, taskID, sessionID, "successor", "", false, nil) + if err != nil { + t.Fatalf("successor steer errored: %v", err) + } + if result == nil || result.StopReason != steerQueuedStopReason { + t.Fatalf("successor result = %+v, want queued steer", result) + } + if status := svc.messageQueue.GetStatus(ctx, sessionID); status == nil || status.Count != 1 { + t.Fatalf("queue before predecessor failure = %+v, want one successor", status) + } + + // The predecessor turn ends while the steer still owns the admission slot. + if err := repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("set session promptable: %v", err) + } + if svc.drainQueuedMessageForPromptableSession(ctx, sessionID) { + t.Fatal("completion drain dispatched while the steer was still in flight") + } + + close(steerRelease) + if err := <-firstDone; err == nil { + t.Fatal("predecessor steer failure was not returned") + } + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if status := svc.messageQueue.GetStatus(ctx, sessionID); status != nil && status.Count == 0 { + select { + case <-promptDone: + return + case <-time.After(2 * time.Second): + t.Fatal("queued successor was removed but its prompt did not dispatch") + } + } + time.Sleep(5 * time.Millisecond) + } + if status := svc.messageQueue.GetStatus(ctx, sessionID); status == nil || status.Count != 0 { + t.Fatalf("successor remained queued after pre-dispatch failure: %+v", status) + } } // TestSteerTask_NotEligibleReturnsSentinel pins that an ineligible session yields diff --git a/apps/web/e2e/helpers/generating-session.ts b/apps/web/e2e/helpers/generating-session.ts new file mode 100644 index 0000000000..d84316b85d --- /dev/null +++ b/apps/web/e2e/helpers/generating-session.ts @@ -0,0 +1,52 @@ +import { type Page } from "@playwright/test"; +import { expect } from "../fixtures/test-base"; +import type { SeedData } from "../fixtures/test-base"; +import type { ApiClient } from "./api-client"; +import { waitForActiveSessionForegroundActivity } from "./session-store"; +import { SessionPage } from "../pages/session-page"; + +export interface SeedRunningGeneratingSessionOptions { + // Duration for the default `/sleep N` predecessor. Ignored when + // predecessorPrompt is set. + sleepSeconds?: number; + // Overrides the default `/sleep N` predecessor entirely — used by the + // folded/deferred tests to seed steer-fold-setup / steer-defer-setup + // instead. + predecessorPrompt?: string; +} + +/** + * Seeds a session whose foreground turn is still generating, which is the state + * that normally queues further input. A no-tool `/sleep` holds the turn open + * without emitting output, so a test can send into a genuinely busy session + * without racing agent activity. + */ +export async function seedRunningGeneratingSession( + testPage: Page, + apiClient: ApiClient, + seedData: SeedData, + title: string, + options: SeedRunningGeneratingSessionOptions = {}, +): Promise<{ session: SessionPage; taskId: string; sessionId: string }> { + const { sleepSeconds = 60, predecessorPrompt } = options; + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + title, + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await session.waitForChatIdle({ timeout: 30_000 }); + await session.sendMessage(predecessorPrompt ?? `/sleep ${sleepSeconds}`); + await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); + await waitForActiveSessionForegroundActivity(testPage, "generating"); + if (!task.session_id) throw new Error("createTaskWithAgent did not return a session_id"); + return { session, taskId: task.id, sessionId: task.session_id }; +} diff --git a/apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts b/apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts new file mode 100644 index 0000000000..9cdc055c23 --- /dev/null +++ b/apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts @@ -0,0 +1,108 @@ +import fs from "node:fs"; +import path from "node:path"; +import { test, expect } from "../../fixtures/test-base"; +import { seedRunningGeneratingSession } from "../../helpers/generating-session"; +import { typeWhileBusy } from "../../helpers/type-while-busy"; + +// Regression coverage for AC-TASKS-PROMPT-ATTACHMENTS-001.6: an attachment sent +// into a turn that is already generating must be materialized into the session +// before the agent is prompted. +// +// The steer route dispatched the prompt without materializing, so the agent +// received a descriptor whose bytes were never written into the session and the +// attachment silently read as empty. The observable proof is the materialized +// file on disk: the UI looks identical either way, because the descriptor is +// persisted on the user message whether or not the bytes ever landed. + +const ATTACHMENT_NAME = "mid-turn-evidence.txt"; + +// Locates the materialized attachment beneath any session's +// `.kandev/attachments//` directory. The ACP session id is not +// exposed to the browser, so the file is found by name rather than by path. +function findMaterializedAttachment(root: string, name: string): string | null { + const stack: string[] = [root]; + while (stack.length > 0) { + const dir = stack.pop(); + if (dir === undefined) continue; + let entries: fs.Dirent[]; + try { + entries = fs.readdirSync(dir, { withFileTypes: true }); + } catch { + continue; // A worktree can disappear under us mid-scan. + } + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + if (entry.name === "node_modules" || entry.name === ".git") continue; + stack.push(full); + } else if (entry.isFile() && entry.name === name && full.includes(".kandev")) { + return full; + } + } + } + return null; +} + +test.describe.serial("mid-turn attachment delivery", () => { + test.describe.configure({ retries: 1 }); + + test.beforeAll(async ({ backend }) => { + await backend.restart({ KANDEV_FEATURES_CLAUDE_MID_TURN_STEERING: "true" }); + const response = await fetch(`${backend.baseUrl}/api/v1/features`); + expect(response.ok).toBeTruthy(); + expect(await response.json()).toMatchObject({ claudeMidTurnSteering: true }); + }); + + test.afterAll(async ({ backend }) => { + await backend.restart(); + }); + + test("sends an attachment into a generating turn", async ({ + testPage, + apiClient, + seedData, + backend, + }, testInfo) => { + test.setTimeout(120_000); + const { session } = await seedRunningGeneratingSession( + testPage, + apiClient, + seedData, + "Mid-turn attachment", + ); + + fs.mkdirSync(testInfo.outputDir, { recursive: true }); + const attachmentPath = path.join(testInfo.outputDir, ATTACHMENT_NAME); + fs.writeFileSync(attachmentPath, "mid-turn attachment bytes"); + + await testPage.locator('input[type="file"]').setInputFiles(attachmentPath); + await expect(testPage.getByText(ATTACHMENT_NAME, { exact: true })).toBeVisible({ + timeout: 10_000, + }); + + const editor = session.activeChat().locator(".tiptap.ProseMirror:visible"); + await typeWhileBusy(testPage, editor, "read the attached file"); + await testPage.getByTestId("submit-message-button").click(); + + // The user message is persisted immediately. The file assertion below + // proves that the claimed bytes reached the session working copy. + await expect( + session.activeChat().getByTestId("user-message-bubble").filter({ + hasText: "read the attached file", + }), + ).toBeVisible({ timeout: 15_000 }); + + // The bytes reached the session. Before the fix this file never existed, + // because the steer route dispatched without materializing. + await expect + .poll(() => findMaterializedAttachment(backend.tmpDir, ATTACHMENT_NAME) !== null, { + timeout: 20_000, + message: "Wait for the attachment to be materialized into the session", + }) + .toBe(true); + + const materialized = findMaterializedAttachment(backend.tmpDir, ATTACHMENT_NAME); + if (materialized === null) throw new Error("Materialized attachment disappeared"); + expect(fs.readFileSync(materialized, "utf8")).toBe("mid-turn attachment bytes"); + }); +}); diff --git a/apps/web/e2e/tests/chat/mid-turn-steering.spec.ts b/apps/web/e2e/tests/chat/mid-turn-steering.spec.ts index 8591fa5363..60978f2e78 100644 --- a/apps/web/e2e/tests/chat/mid-turn-steering.spec.ts +++ b/apps/web/e2e/tests/chat/mid-turn-steering.spec.ts @@ -1,11 +1,8 @@ import { type Page } from "@playwright/test"; import { test, expect } from "../../fixtures/test-base"; -import type { SeedData } from "../../fixtures/test-base"; -import type { ApiClient } from "../../helpers/api-client"; -import { waitForActiveSessionForegroundActivity } from "../../helpers/session-store"; import { typeWhileBusy } from "../../helpers/type-while-busy"; -import { SessionPage } from "../../pages/session-page"; import { registerSeparateQueueRows } from "../../helpers/message-queue-settings"; +import { seedRunningGeneratingSession } from "../../helpers/generating-session"; registerSeparateQueueRows(test); @@ -18,50 +15,6 @@ registerSeparateQueueRows(test); // tiptap placeholder is a cursor-anchored decoration and only renders under // focus, so asserting message delivery is the robust signal. -interface SeedRunningGeneratingSessionOptions { - // Duration for the default `/sleep N` predecessor. Ignored when - // predecessorPrompt is set. - sleepSeconds?: number; - // Overrides the default `/sleep N` predecessor entirely — used by the - // folded/deferred tests to seed steer-fold-setup / steer-defer-setup - // instead. - predecessorPrompt?: string; -} - -async function seedRunningGeneratingSession( - testPage: Page, - apiClient: ApiClient, - seedData: SeedData, - title: string, - options: SeedRunningGeneratingSessionOptions = {}, -): Promise<{ session: SessionPage; taskId: string; sessionId: string }> { - const { sleepSeconds = 20, predecessorPrompt } = options; - const task = await apiClient.createTaskWithAgent( - seedData.workspaceId, - title, - seedData.agentProfileId, - { - description: "/e2e:simple-message", - workflow_id: seedData.workflowId, - workflow_step_id: seedData.startStepId, - repository_ids: [seedData.repositoryId], - }, - ); - await testPage.goto(`/t/${task.id}`); - const session = new SessionPage(testPage); - await session.waitForLoad(); - await session.waitForChatIdle({ timeout: 30_000 }); - // A no-tool sleep holds the foreground turn open and generating — the exact - // state that normally queues input, so it isolates the steering gate. - // steer-fold-setup/steer-defer-setup hold the same way, but differ in - // whether they answer on their own first (see the folded/deferred tests). - await session.sendMessage(predecessorPrompt ?? `/sleep ${sleepSeconds}`); - await expect(session.agentStatus()).toBeVisible({ timeout: 15_000 }); - await waitForActiveSessionForegroundActivity(testPage, "generating"); - if (!task.session_id) throw new Error("createTaskWithAgent did not return a session_id"); - return { session, taskId: task.id, sessionId: task.session_id }; -} - test.describe.serial("Claude mid-turn steering experiment", () => { test.describe.configure({ retries: 1 }); diff --git a/docs/plans/steer-attachment-materialization/plan.md b/docs/plans/steer-attachment-materialization/plan.md new file mode 100644 index 0000000000..4d2ebb0504 --- /dev/null +++ b/docs/plans/steer-attachment-materialization/plan.md @@ -0,0 +1,174 @@ +--- +created: 2026-09-02 +status: implemented +requirements: + - REQ-TASKS-PROMPT-ATTACHMENTS-001 +system_design: + - ../../specs/tasks/system-design/prompt-attachments.md +legacy_specs: [] +--- + +# Implementation Plan: Steer Attachment Materialization + +## Overview + +A message sent while the agent is still generating is delivered through the +steer route. That route dispatches the prompt without materializing the claimed +attachment descriptors it carries. + +The agent therefore receives a bounded descriptor whose bytes were never +written to `.kandev/attachments//`. The reference resolves to +nothing and the attachment reads as empty. + +The failure is silent. The steer route has no equivalent of the ordinary +prompt path's materialization error, so the user sees a delivered message and +an agent that cannot see the file. + +This plan moves materialization ahead of the steer dispatch, keeps it outside +the prompt lifecycle lock, and makes a materialization failure terminal for +that message instead of silently degrading. + +## Scope + +### In scope + +- Materialize claimed descriptors before a steer dispatch. +- Route a lost steer race through ordinary prompt admission. +- Return a durable error when materialization fails on the steer route. +- Release the steer slot and retry queued draining after a pre-dispatch failure. +- Backend regression coverage in the lifecycle package. +- One Playwright scenario for a mid-turn attachment. + +### Out of scope + +- Upload limits, staging, claim admission, and retention. +- Delivery-mode selection and the prompt composer. +- The launch and initial-prompt routes fixed by + `docs/plans/session-launch-prompt-admission/`. +- Queued-message delivery, which already reaches the ordinary prompt path. + +## Reproduction evidence + +The defect was reproduced on a live instance. The upload succeeded and the +backend blob was written, but the session's +`.kandev/attachments//` directory stayed empty and the agent +received a resource reference that resolved to nothing. + +Two observations look like counter-evidence and are not: + +- `KANDEV_FEATURES_CLAUDE_MID_TURN_STEERING=false` appears in the backend + process environment. `profiles.ApplyProfile` stamps profile defaults into the + process with `os.Setenv` and records them, and `runtimeflags/config.go` treats + an env var as explicit only when `!profiles.WasApplied(name)`. The + `runtime_flag_overrides` row `features.claudeMidTurnSteering = 1` therefore + wins and steering is enabled. +- The backend log shows no `sending prompt to agent` line carrying the + attachment. That line is emitted from `preparePrompt`, which only `sendPrompt` + calls. The steer route never reaches it, so its absence is what this defect + predicts rather than evidence against it. + +## Root cause + +`SessionManager.sendPrompt` in +`apps/backend/internal/agent/runtime/lifecycle/session.go` calls +`sm.materializeAttachments` before `sm.triggerPrompt`. + +`SessionManager.dispatchSteerLocked` in the same file calls +`sm.callAgentctlPrompt` with the caller's `attachments` slice untouched. No +materialization runs on that route. + +`SendPromptSteerWithDispatchCallback` tries the steer first. When a turn is +live the steer succeeds and the unmaterialized descriptors reach the agent. +When no turn is live it falls back to `sendPrompt`, which materializes +correctly. That is why the defect only appears mid-turn. + +## Technical approach + +### Materialize before the steer + +Resolve descriptors in `SendPromptSteerWithDispatchCallback`, after the status +check and before `tryDispatchSteer`. + +Materialization streams up to 100 MiB over HTTP. It must not run inside +`dispatchSteerLocked`, which holds `execution.promptLifecycleMu` under +`steerDispatchTimeout`. Resolving in the caller keeps the lock bounded and +preserves the existing steer latency contract. + +Return the materialization error to the caller. Do not dispatch and do not fall +back to an ordinary prompt, because the fallback would deliver the same +unresolved descriptors. + +### Return lost races to ordinary admission + +Check for an active generation before uploading. If the generation ends during +materialization, return a typed no-dispatch error instead of calling +`sendPrompt` directly. The orchestrator maps this result to its existing +ordinary prompt fallback, so foreground claims and prompt-cycle state stay in +one owner. + +Wrap materialization failures with a separate pre-dispatch identity. The +orchestrator releases `steerInFlight` and retries the queued-message drain +after this failure, so a completion that raced with the upload cannot strand a +successor. + +### Regression coverage + +`newMockAgentServer` in +`apps/backend/internal/agent/runtime/lifecycle/session_test.go` already serves +an `http.ServeMux`. Add an `/api/v1/attachments/materialize` route that records +each upload, so `steerHarness` can assert materialization on the wire. + +The New Agent mid-turn scenario belongs with the existing chat attachment +specs in `apps/web/e2e/tests/chat/`. + +## Tests + +| Acceptance criterion | Evidence | +| --- | --- | +| `AC-TASKS-PROMPT-ATTACHMENTS-001.6` | `TestSendPromptSteer_MaterializesAttachmentsBeforeDispatch` in `apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go` | +| `AC-TASKS-PROMPT-ATTACHMENTS-001.6` | `TestSendPrompt_MaterializesAttachmentsExactlyOnce` in `apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go` | +| `AC-TASKS-PROMPT-ATTACHMENTS-001.7` | `TestSendPromptSteer_MaterializationFailureDoesNotDispatch` in `apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go` | +| `AC-TASKS-PROMPT-ATTACHMENTS-001.7` | `TestSendPromptSteer_DoesNotFallbackOutsideAdmission` and `TestSteerTask_PreDispatchFailureDrainsSuccessor` cover lost-generation admission and queue recovery. | +| `AC-TASKS-PROMPT-ATTACHMENTS-001.6` | `sends an attachment into a generating turn` in `apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts` | + +## E2E tests + +`apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts` will contain +`sends an attachment into a generating turn`. + +The scenario starts a turn, uploads a file while that turn is still generating, +sends it, and asserts the agent resolves the file contents. It must fail before +the correction because the materialized file is absent from the session +directory. + +## Work orders + +- [x] [Task 01: Materialize steer attachments](task-01-materialize-steer-attachments.md) +- [x] [Task 02: Cover mid-turn attachment delivery](task-02-mid-turn-attachment-e2e.md) + +## Verification results + +- The three steer regression tests failed first for the expected reasons: zero + materialization uploads, and no error raised on a failed materialization. +- They pass after the correction, including under the race detector. +- The full `internal/agent/runtime/lifecycle` package passes. +- `golangci-lint run ./...` reports 0 issues and `gofmt` reports no files. +- The web typecheck passes. +- `tests/chat/mid-turn-attachment.spec.ts` failed against the reverted backend + on the materialization assertion, and passes with the correction. +- The six pre-existing `mid-turn-steering.spec.ts` scenarios still pass after + `seedRunningGeneratingSession` moved to a shared helper. + +## Risks + +- `promptLifecycleMu` bounds steer latency. Materializing inside it would let a + large upload block the lifecycle lock for the length of a network write. +- The steer route can lose its race while an upload is in progress. The + orchestrator must own the replacement prompt and its admission checks. +- `materializeAttachments` requires a non-empty `ACPSessionID`. A steer implies + a live turn, but the error must stay a returned error rather than a panic. +- Making a steer materialization failure terminal changes an existing silent + path into a visible one. Queued and ordinary prompts must keep their current + ordering behavior. +- Playwright timing for a mid-turn send depends on a turn that is still + generating. The scenario needs a deterministic long-running mock turn. diff --git a/docs/plans/steer-attachment-materialization/task-01-materialize-steer-attachments.md b/docs/plans/steer-attachment-materialization/task-01-materialize-steer-attachments.md new file mode 100644 index 0000000000..345837275d --- /dev/null +++ b/docs/plans/steer-attachment-materialization/task-01-materialize-steer-attachments.md @@ -0,0 +1,62 @@ +--- +id: "01-materialize-steer-attachments" +title: "Materialize steer attachments" +status: complete +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-PROMPT-ATTACHMENTS-001 +acceptance_criteria: + - AC-TASKS-PROMPT-ATTACHMENTS-001.6 + - AC-TASKS-PROMPT-ATTACHMENTS-001.7 +system_design: + - ../../specs/tasks/system-design/prompt-attachments.md +--- + +# Task 01: Materialize Steer Attachments + +## Summary + +Resolve claimed attachment descriptors before a steer dispatches into a live +turn. Keep the resolution outside the prompt lifecycle lock and make a +resolution failure terminal for that message. + +## In scope + +- Materialize in `SendPromptSteerWithDispatchCallback` before `tryDispatchSteer`. +- Return a no-dispatch result when the active generation ends before steer + delivery, so the orchestrator owns the ordinary prompt fallback. +- Release the steer admission slot and retry queued draining after a + pre-dispatch materialization failure. +- Return the materialization error instead of dispatching or falling back. +- Record materialize uploads in `newMockAgentServer` for assertions. +- Add the three lifecycle regression tests named in the plan. + +## Out of scope + +- Claim admission, upload limits, retention, and delivery-mode selection. +- The Playwright scenario, which is Task 02. + +## Acceptance + +- A steer into a generating turn reaches agentctl with a resolved filename and + no bare attachment ID. +- The ordinary prompt route uploads attachment bytes exactly once. +- A materialization failure returns an error and dispatches no prompt frame. +- Existing steer generation and ordering behavior is unchanged. + +## Verification + +```bash +gofmt -l apps/backend/internal/agent/runtime/lifecycle +(cd apps/backend && go test ./internal/agent/runtime/lifecycle -run 'TestSendPromptSteer' -count=1 -race) +(cd apps/backend && go test ./internal/agent/runtime/lifecycle -count=1) +make -C apps/backend lint +``` + +## Files likely touched + +- `apps/backend/internal/agent/runtime/lifecycle/session.go` +- `apps/backend/internal/agent/runtime/lifecycle/session_steer_test.go` +- `apps/backend/internal/agent/runtime/lifecycle/session_test.go` diff --git a/docs/plans/steer-attachment-materialization/task-02-mid-turn-attachment-e2e.md b/docs/plans/steer-attachment-materialization/task-02-mid-turn-attachment-e2e.md new file mode 100644 index 0000000000..787a2a6077 --- /dev/null +++ b/docs/plans/steer-attachment-materialization/task-02-mid-turn-attachment-e2e.md @@ -0,0 +1,55 @@ +--- +id: "02-mid-turn-attachment-e2e" +title: "Cover mid-turn attachment delivery" +status: complete +wave: 1 +depends_on: + - "01-materialize-steer-attachments" +plan: "plan.md" +requirements: + - REQ-TASKS-PROMPT-ATTACHMENTS-001 +acceptance_criteria: + - AC-TASKS-PROMPT-ATTACHMENTS-001.6 +system_design: + - ../../specs/tasks/system-design/prompt-attachments.md +--- + +# Task 02: Cover Mid-Turn Attachment Delivery + +## Summary + +Add one Playwright scenario that uploads a file while a turn is still +generating and asserts the agent receives the file contents. + +## In scope + +- Add `apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts`. +- Start a turn that keeps generating long enough to send a second message. +- Upload and send an attachment during that turn. +- Assert the agent resolves the file contents, not an empty reference. + +## Out of scope + +- Backend changes, which are Task 01. +- Queued-message delivery and delivery-mode coverage, already covered by + `apps/web/e2e/tests/chat/attachment-delivery-mode.spec.ts`. + +## Acceptance + +- The scenario fails before Task 01 because the attachment resolves empty. +- The scenario passes after Task 01. +- The shared helper keeps the mock turn open for 60 seconds by default, which + leaves setup headroom under CI load before the steer is submitted. + +## Verification + +```bash +(cd apps/web && pnpm run lint) +make -C apps/backend build +(cd apps/web && pnpm run build:e2e) +(cd apps/web && pnpm e2e:run tests/chat/mid-turn-attachment.spec.ts) +``` + +## Files likely touched + +- `apps/web/e2e/tests/chat/mid-turn-attachment.spec.ts` diff --git a/docs/specs/tasks/requirements/prompt-attachments.md b/docs/specs/tasks/requirements/prompt-attachments.md index f587cb8a66..c15b0f8b4f 100644 --- a/docs/specs/tasks/requirements/prompt-attachments.md +++ b/docs/specs/tasks/requirements/prompt-attachments.md @@ -2,7 +2,7 @@ status: draft system: tasks created: 2026-08-04 -updated: 2026-09-01 +updated: 2026-09-02 owners: - Kandev team --- @@ -33,6 +33,14 @@ This document is the migrated task-system source for the capability. The source - **AC-TASKS-PROMPT-ATTACHMENTS-001.5:** When attachment materialization fails before the initial prompt reaches the agent, the system shall show a durable launch error and shall not present the session as active work. +- **AC-TASKS-PROMPT-ATTACHMENTS-001.6:** When a message carrying claimed + attachments is delivered into a turn that is already generating, the system + shall materialize those attachments into the active session before the agent + receives the message, on the same terms as a message that starts a new turn. +- **AC-TASKS-PROMPT-ATTACHMENTS-001.7:** When materialization fails for a + message delivered into a generating turn, the system shall surface a durable + error for that message and shall not deliver an attachment reference whose + bytes are absent from the session. ## Migrated source detail @@ -74,6 +82,8 @@ move files into a workspace manually or strip useful evidence. - Path-delivered files are materialized in the active execution beneath the session's `.kandev/attachments//` directory before the agent is prompted. Existing native-prompt and path delivery choices remain available. + This holds for every delivery route, including a message folded into a turn + that is already generating. - Desktop and mobile expose the same upload, retry, removal, submission, and size-error outcomes through their existing prompt composers. diff --git a/docs/specs/tasks/system-design/prompt-attachments.md b/docs/specs/tasks/system-design/prompt-attachments.md index 2416a88ad7..a05050b247 100644 --- a/docs/specs/tasks/system-design/prompt-attachments.md +++ b/docs/specs/tasks/system-design/prompt-attachments.md @@ -70,6 +70,13 @@ The lifecycle manager starts prompt generation before materialization. A materialization error therefore uses the same terminal prompt-error path as an ACP submission error. +Delivery into a turn that is already generating uses the same materialization +step. The steer route resolves descriptors before it acquires the prompt +lifecycle lock, because materialization streams file bytes over the network and +the lock is held only for a bounded dispatch. A steer whose materialization +fails does not dispatch and does not fall through to an ordinary prompt +carrying unresolved descriptors. + ## Failure and recovery A claim error returns from `session.launch` before the intent changes runtime