fix(agent): materialize attachments sent into a generating turn (#3277)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
@@ -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/<acp-session-id>/` 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");
|
||||
});
|
||||
});
|
||||
@@ -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 });
|
||||
|
||||
|
||||
@@ -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/<acp-session-id>/`. 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/<acp-session-id>/` 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.
|
||||
@@ -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`
|
||||
@@ -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`
|
||||
@@ -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/<session-id>/` 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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user