//go:build unix
// Acknowledgement loss must not duplicate effects in the real daemon. The
// proxy waits for admission before cutting the client response, so these
// scenarios cannot accidentally test a failure that preceded execution.
package e2e
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"maps"
"net"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"strconv"
"strings"
"sync"
"sync/atomic"
tea "charm.land/bubbletea/v2"
"testing"
"time"
"albedo/cli/internal/app"
"albedo/cli/internal/daemon"
"albedo/cli/internal/tui"
)
type acknowledgementFault string
const (
expiredAcknowledgement acknowledgementFault = "expired"
dropAcknowledgement acknowledgementFault = "drop"
hideReceiptAcknowledgement acknowledgementFault = "drop-and-hide-receipt"
unresolvedAcknowledgement acknowledgementFault = "unresolved"
truncateAcknowledgement acknowledgementFault = "truncate"
rejectAcknowledgement acknowledgementFault = "server-error"
cancelAcknowledgement acknowledgementFault = "cancel"
emptyAcknowledgement acknowledgementFault = "empty"
malformedAcknowledgement acknowledgementFault = "malformed"
missingAcknowledgement acknowledgementFault = "missing-fields"
negativeAcknowledgement acknowledgementFault = "negative"
wrongStatusAcknowledgement acknowledgementFault = "wrong-status"
)
type acknowledgementProxy struct {
expireReceipt atomic.Bool
allowReceipt atomic.Bool
failure error
connection *daemon.Connection
admitted chan struct{}
responses [][]byte
requests []string
mu sync.Mutex
}
func cutAcknowledgement(t *testing.T, path string, fault acknowledgementFault) *acknowledgementProxy {
t.Helper()
return mutateAcknowledgement(t, path, fault, nil, nil)
}
func mutateAcknowledgement(t *testing.T, path string, fault acknowledgementFault, matches func([]byte) bool, transform func([]byte) ([]byte, error)) *acknowledgementProxy {
t.Helper()
upstream := conn(t).Snapshot()
destination, err := url.Parse(conn(t).BaseURL())
if err != nil {
t.Fatal(err)
}
proxy := &acknowledgementProxy{admitted: make(chan struct{}, 1)}
transport := &http.Transport{}
t.Cleanup(transport.CloseIdleConnections)
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
requestBody, bodyErr := io.ReadAll(request.Body)
if bodyErr != nil {
http.Error(writer, bodyErr.Error(), http.StatusBadRequest)
return
}
if fault == unresolvedAcknowledgement && request.Method == http.MethodGet && strings.HasPrefix(request.URL.Path, "/operations/") && !proxy.allowReceipt.Load() {
writer.WriteHeader(http.StatusServiceUnavailable)
_, _ = writer.Write([]byte(`{"error":"receipt unavailable"}`))
return
}
proxy.mu.Lock()
hideReceipt := fault == hideReceiptAcknowledgement && request.Method == http.MethodGet && strings.HasPrefix(request.URL.Path, "/operations/") && !slices.ContainsFunc(proxy.requests, func(path string) bool { return strings.HasPrefix(path, "GET /operations/") })
proxy.requests = append(proxy.requests, request.Method+" "+request.URL.Path)
proxy.mu.Unlock()
if request.Method == http.MethodGet && strings.HasPrefix(request.URL.Path, "/operations/") && (fault == expiredAcknowledgement || proxy.expireReceipt.Load()) {
writer.WriteHeader(http.StatusGone)
_, _ = writer.Write([]byte(`{"code":"operation_expired","error":"expired"}`))
return
}
if hideReceipt {
writer.WriteHeader(http.StatusNotFound)
_, _ = writer.Write([]byte(`{"code":"operation_unknown","error":"unknown"}`))
return
}
forwarded := request.Clone(request.Context())
forwarded.Body = http.NoBody
if len(requestBody) > 0 {
forwarded.Body = io.NopCloser(bytes.NewReader(requestBody))
}
forwarded.ContentLength = int64(len(requestBody))
forwarded.TransferEncoding = nil
forwarded.URL.Scheme, forwarded.URL.Host = destination.Scheme, destination.Host
forwarded.RequestURI = ""
forwarded.Host = destination.Host
response, forwardErr := transport.RoundTrip(forwarded)
if forwardErr != nil {
proxy.mu.Lock()
proxy.failure = forwardErr
proxy.mu.Unlock()
http.Error(writer, forwardErr.Error(), http.StatusBadGateway)
return
}
defer response.Body.Close()
if strings.HasPrefix(response.Header.Get("Content-Type"), "text/event-stream") {
for key, values := range response.Header {
writer.Header()[key] = values
}
writer.WriteHeader(response.StatusCode)
_ = http.NewResponseController(writer).Flush()
_, _ = io.Copy(flushingResponseWriter{writer}, response.Body)
return
}
body, readErr := io.ReadAll(io.LimitReader(response.Body, 1<<20))
if readErr != nil {
proxy.mu.Lock()
proxy.failure = readErr
proxy.mu.Unlock()
http.Error(writer, readErr.Error(), http.StatusBadGateway)
return
}
isMutation := request.Method == http.MethodPost && request.URL.Path == path && (matches == nil || matches(requestBody))
first := false
if isMutation {
proxy.mu.Lock()
first = len(proxy.responses) == 0
proxy.responses = append(proxy.responses, body)
if response.StatusCode < 200 || response.StatusCode >= 300 {
proxy.failure = fmt.Errorf("daemon rejected mutation: %s: %s", response.Status, body)
}
proxy.mu.Unlock()
}
if isMutation && (first || fault == unresolvedAcknowledgement || fault == expiredAcknowledgement) {
if first {
proxy.admitted <- struct{}{}
}
if transform != nil {
transformed, transformErr := transform(body)
if transformErr != nil {
proxy.mu.Lock()
proxy.failure = transformErr
proxy.mu.Unlock()
http.Error(writer, transformErr.Error(), http.StatusBadGateway)
return
}
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write(transformed)
return
}
switch fault {
case dropAcknowledgement, hideReceiptAcknowledgement, unresolvedAcknowledgement, expiredAcknowledgement:
socket, _, hijackErr := writer.(http.Hijacker).Hijack()
if hijackErr == nil {
_ = socket.Close()
}
return
case truncateAcknowledgement:
writer.Header().Set("Content-Length", strconv.Itoa(len(body)+1))
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write(body[:len(body)/2])
return
case rejectAcknowledgement:
http.Error(writer, "acknowledgement unavailable", http.StatusServiceUnavailable)
return
case emptyAcknowledgement:
writer.WriteHeader(response.StatusCode)
return
case malformedAcknowledgement:
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write([]byte("{"))
return
case missingAcknowledgement:
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write([]byte("{}"))
return
case negativeAcknowledgement:
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write([]byte(`{"ok":false,"queued":false,"submitted":false}`))
return
case wrongStatusAcknowledgement:
writer.WriteHeader(http.StatusCreated)
_, _ = writer.Write(body)
return
case cancelAcknowledgement:
<-request.Context().Done()
return
}
}
maps.Copy(writer.Header(), response.Header)
writer.WriteHeader(response.StatusCode)
_, _ = writer.Write(body)
}))
t.Cleanup(server.Close)
upstream.Port = server.Listener.Addr().(*net.TCPAddr).Port
proxy.connection = daemon.NewConnection(upstream, func(context.Context) (daemon.ConnectionSnapshot, error) { return upstream, nil })
t.Cleanup(proxy.connection.HTTPClient().CloseIdleConnections)
return proxy
}
func (proxy *acknowledgementProxy) assertSingleEffect(t *testing.T, operationErr error) []byte {
t.Helper()
if _, uncertain := errors.AsType[*daemon.UncertainOutcomeError](operationErr); !uncertain {
t.Fatalf("lost acknowledgement should be uncertain, got %v", operationErr)
}
return proxy.acceptedResponse(t)
}
func (proxy *acknowledgementProxy) acceptedResponse(t *testing.T) []byte {
t.Helper()
proxy.mu.Lock()
defer proxy.mu.Unlock()
if proxy.failure != nil {
t.Fatal(proxy.failure)
}
if len(proxy.responses) != 1 {
t.Fatalf("daemon admitted %d mutations, want one", len(proxy.responses))
}
return slices.Clone(proxy.responses[0])
}
func TestLostCreationAcknowledgementCreatesOneSession(t *testing.T) {
providerRoute(t, echoReply)
for _, fault := range []acknowledgementFault{dropAcknowledgement, truncateAcknowledgement, rejectAcknowledgement, emptyAcknowledgement, malformedAcknowledgement, missingAcknowledgement} {
t.Run(string(fault), func(t *testing.T) {
workspace := t.TempDir()
proxy := cutAcknowledgement(t, "/sessions", fault)
service := app.Service{Connect: func(context.Context) (*daemon.Connection, error) { return proxy.connection, nil }}
opened, err := service.PrepareOpen(context.Background(), app.OpenOptions{Workspace: workspace, Fresh: true})
if err != nil {
t.Fatal(err)
}
accepted := proxy.acceptedResponse(t)
var created daemon.Session
if decodeErr := json.Unmarshal(accepted, &created); decodeErr != nil || created.ID == "" {
t.Fatalf("accepted session: %s, %v", accepted, decodeErr)
}
if opened.Selected == nil || opened.Selected.ID != created.ID {
t.Fatal("lost creation receipt was not recovered")
}
sessions, err := daemon.ListSessions(context.Background(), conn(t))
if err != nil {
t.Fatal(err)
}
count := 0
for _, session := range sessions {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("workspace contains %d sessions, want one", count)
}
})
}
}
func TestRejectedCreationReceiptRecoversWithoutAllocatingSession(t *testing.T) {
t.Parallel()
workspace := t.TempDir()
handle, err := daemon.NewCreation(daemon.CreateSessionRequest{Workspace: workspace, Provider: "missing-creation-profile"})
if err != nil {
t.Fatal(err)
}
_, err = daemon.CreateSessionOperation(t.Context(), conn(t), handle)
initial, ok := errors.AsType[*daemon.APIError](err)
if !ok || initial.StatusCode != http.StatusBadRequest {
t.Fatalf("unknown provider did not reject creation: %v", err)
}
// A rejected creation has a durable receipt but no session target. An
// independent receipt read must recover that rejection without treating
// the empty target as malformed or submitting another creation.
receipt, err := daemon.ResolveOperation(t.Context(), conn(t), handle)
if err != nil || receipt.Status != "rejected" || receipt.Target != "" || receipt.OperationID != handle.ID() {
t.Fatalf("rejected creation receipt: %+v, %v", receipt, err)
}
created, err := daemon.ResolveCreation(t.Context(), conn(t), handle)
recovered, ok := errors.AsType[*daemon.APIError](err)
if !ok || recovered.StatusCode != initial.StatusCode || created.ID != "" {
t.Fatalf("resolve rejected creation: %+v, %v", created, err)
}
sessions, err := daemon.ListSessions(t.Context(), conn(t))
if err != nil {
t.Fatal(err)
}
for _, session := range sessions {
if session.Workspace == workspace {
t.Fatal("rejected creation allocated a session")
}
}
}
func TestInvalidTUICreationAcknowledgementRecoversCreatedSession(t *testing.T) {
profile := providerRoute(t, echoReply)
for _, fault := range []acknowledgementFault{dropAcknowledgement, emptyAcknowledgement, malformedAcknowledgement, missingAcknowledgement} {
t.Run(string(fault), func(t *testing.T) {
initial := daemonSession(t, newFaultSession(t, profile))
workspace := t.TempDir()
proxy := cutAcknowledgement(t, "/sessions", fault)
driver := driveTUIWithConnection(t, &initial, proxy.connection)
defer driver.App.Chat.Close()
result := driver.Send(tui.FolderNewSessionMsg{Workspace: workspace})
driver.Update(result)
accepted := proxy.acceptedResponse(t)
var created daemon.Session
if decodeErr := json.Unmarshal(accepted, &created); decodeErr != nil || created.ID == "" {
t.Fatalf("accepted session: %s, %v", accepted, decodeErr)
}
if driver.App.ActiveSession == nil || driver.App.ActiveSession.ID != created.ID || driver.App.Chat.Notices.HasError() {
t.Fatalf("lost creation response did not recover its created session: %+v", driver.App.Chat.Notices)
}
count := 0
for _, session := range daemonSessions(t) {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("TUI creation admitted %d sessions, want one", count)
}
})
}
}
func TestInvalidSubmissionAcknowledgementSendsOneTurn(t *testing.T) {
for _, fault := range []acknowledgementFault{dropAcknowledgement, emptyAcknowledgement, malformedAcknowledgement, missingAcknowledgement, negativeAcknowledgement, wrongStatusAcknowledgement} {
t.Run(string(fault), func(t *testing.T) {
profile := "invalid-submit-" + string(fault)
if err := suite.provider.addProfile(profile, echoReply); err != nil {
t.Fatal(err)
}
id := newFaultSession(t, profile)
proxy := cutAcknowledgement(t, "/sessions/"+id+"/events", fault)
client := daemon.NewChatClient(proxy.connection, id)
result, err := client.Send(context.Background(), "exactly one uncertain turn", nil)
if err != nil || result == nil || result.OperationID == "" {
t.Fatalf("recover submission: %+v %v", result, err)
}
proxy.acceptedResponse(t)
waitIdle(t, id, profile, 1)
if count := len(suite.provider.requests(profile)); count != 1 {
t.Fatalf("provider received %d turns, want one", count)
}
history, err := daemon.NewChatClient(conn(t), id).History(context.Background(), 0, 120)
if err != nil {
t.Fatal(err)
}
count := 0
for _, event := range history.Events {
if event.Type == daemon.EventUser {
count++
}
}
if count != 1 {
t.Fatalf("transcript contains %d user submissions, want one", count)
}
})
}
}
func TestInvalidCommandAcknowledgementAddsOneWorkItem(t *testing.T) {
profile := providerRoute(t, echoReply)
for _, fault := range []acknowledgementFault{truncateAcknowledgement, emptyAcknowledgement, malformedAcknowledgement, missingAcknowledgement, negativeAcknowledgement, wrongStatusAcknowledgement} {
t.Run(string(fault), func(t *testing.T) {
id := newFaultSession(t, profile)
path := "/sessions/" + id + "/commands"
proxy := cutAcknowledgement(t, path, fault)
session := daemonSession(t, id)
driver := driveTUIWithConnection(t, &session, proxy.connection)
defer driver.App.Chat.Close()
driver.Dispatch(tui.ChatExecuteCommandMsg{Name: "/work", Args: "add one uncertain work item"})
proxy.acceptedResponse(t)
if driver.App.ActiveSession == nil || driver.App.ActiveSession.ID != id || !driver.App.Chat.Notices.HasError() {
t.Fatalf("command acknowledgement loss changed the active session or hid its error: %+v", driver.App.Chat.Notices)
}
listed, err := daemon.LoadPage(context.Background(), conn(t), id, "/work")
if err != nil {
t.Fatal(err)
}
if len(listed.Rows) != 1 || listed.Rows[0].Text != "one uncertain work item" {
t.Fatalf("ledger after lost acknowledgement: %+v", listed)
}
})
}
}
func TestCancellationAfterAdmissionPreservesUncertainCause(t *testing.T) {
profile := providerRoute(t, echoReply)
proxy := cutAcknowledgement(t, "/sessions", cancelAcknowledgement)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
result := make(chan error, 1)
workspace := t.TempDir()
go func() {
_, err := daemon.CreateSession(ctx, proxy.connection, daemon.CreateSessionRequest{Workspace: workspace, Provider: profile})
result <- err
}()
select {
case <-proxy.admitted:
cancel()
case <-time.After(20 * time.Second):
t.Fatal("mutation was not admitted")
}
select {
case err := <-result:
proxy.assertSingleEffect(t, err)
if !errors.Is(err, context.Canceled) {
t.Fatalf("uncertain error lost cancellation cause: %v", err)
}
uncertain, ok := errors.AsType[*daemon.UncertainOutcomeError](err)
if !ok || uncertain.Handle == nil {
t.Fatal("cancellation lost the operation handle")
}
original, resolveErr := daemon.ResolveCreation(t.Context(), proxy.connection, uncertain.Handle)
if resolveErr != nil || original.ID == "" {
t.Fatalf("resolve original creation: %+v %v", original, resolveErr)
}
proxy.acceptedResponse(t)
case <-time.After(5 * time.Second):
t.Fatal("canceled operation did not return")
}
}
func TestLostSignInAcknowledgementStartsOneLocalFlow(t *testing.T) {
proxy := cutAcknowledgement(t, "/auth/antigravity", dropAcknowledgement)
_, err := daemon.StartSignIn(context.Background(), proxy.connection, "antigravity")
accepted := proxy.assertSingleEffect(t, err)
var started daemon.StartedSignIn
if decodeErr := json.Unmarshal(accepted, &started); decodeErr != nil || started.ID == "" {
t.Fatalf("accepted sign-in: %s, %v", accepted, decodeErr)
}
t.Cleanup(func() {
if cancelErr := daemon.CancelSignIn(context.Background(), conn(t), started.ID); cancelErr != nil {
t.Errorf("cancel sign-in: %v", cancelErr)
}
})
status, err := daemon.PollSignIn(context.Background(), conn(t), started.ID)
if err != nil || status.State != "waiting" {
t.Fatalf("the one admitted local flow: %+v, %v", status, err)
}
// Starting only allocates a loopback listener and builds an authorization
// URL. No callback or pasted code is supplied, so no exchange reaches a provider.
}
func TestCoreAuthenticationRefusalRecoversBeforeCreation(t *testing.T) {
profile := providerRoute(t, echoReply)
snapshot := conn(t).Snapshot()
snapshot.Token = "expired-client-token"
stale := daemon.NewConnection(snapshot, func(ctx context.Context) (daemon.ConnectionSnapshot, error) {
return daemon.Rediscover(ctx, suite.home)
})
t.Cleanup(stale.HTTPClient().CloseIdleConnections)
workspace := t.TempDir()
created, err := daemon.CreateSession(context.Background(), stale, daemon.CreateSessionRequest{Workspace: workspace, Provider: profile})
if err != nil || created.ID == "" {
t.Fatalf("creation after pre-admission refusal: %+v, %v", created, err)
}
if stale.Token() != conn(t).Token() {
t.Fatal("recovery did not update credentials")
}
sessions, err := daemon.ListSessions(context.Background(), conn(t))
if err != nil {
t.Fatal(err)
}
count := 0
for _, session := range sessions {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("authentication recovery created %d sessions, want one", count)
}
}
func newFaultSession(t *testing.T, profile string) string {
t.Helper()
created, err := daemon.CreateSession(t.Context(), conn(t), daemon.CreateSessionRequest{Workspace: t.TempDir(), Provider: profile})
if err != nil {
t.Fatal(err)
}
return created.ID
}
func TestWebhookCreationMissingGeneratedSecretStopsAfterOneEffect(t *testing.T) {
profile := providerRoute(t, echoReply)
id := newFaultSession(t, profile)
if _, err := daemon.SelectExtension(t.Context(), conn(t), id, daemon.ExtensionSelectionRequest{Name: "webhooks", Enabled: new(true), Scope: "session"}); err != nil {
t.Fatal(err)
}
path := "/sessions/" + id + "/commands"
proxy := mutateAcknowledgement(t, path, "", func(body []byte) bool {
var command struct {
Args map[string]string `json:"args"`
Name string `json:"name"`
}
return json.Unmarshal(body, &command) == nil && command.Name == "/webhooks" && command.Args["action"] == "create_in"
}, func(body []byte) ([]byte, error) {
var envelope map[string]json.RawMessage
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, err
}
var result map[string]json.RawMessage
if err := json.Unmarshal(envelope["result"], &result); err != nil {
return nil, err
}
if _, present := result["secret"]; !present {
return nil, errors.New("daemon did not generate a secret")
}
delete(result, "secret")
encoded, err := json.Marshal(result)
if err != nil {
return nil, err
}
envelope["result"] = encoded
return json.Marshal(envelope)
})
session := daemonSession(t, id)
driver := driveTUIWithConnection(t, &session, proxy.connection)
defer driver.App.Chat.Close()
driver.Dispatch(tui.ChatOpenWebhooksPageMsg{})
if !driver.App.WebhooksPage.Loaded || driver.App.WebhooksPage.Error != "" {
t.Fatalf("open creation screen: %s", driver.App.WebhooksPage.Error)
}
driver.Dispatch(tea.KeyPressMsg{Code: 'n', Text: "n"})
driver.Type("missing-secret")
// A changed signature adds a second mutation, which must not run after the lost secret.
driver.App.WebhooksPage.Form.Inputs["header"].SetValue("x-custom-signature")
proxy.mu.Lock()
before := len(proxy.requests)
proxy.mu.Unlock()
driver.Dispatch(tea.KeyPressMsg{Code: 's', Mod: tea.ModCtrl})
accepted := proxy.acceptedResponse(t)
page := driver.App.WebhooksPage
if page.Saving || page.Loading || page.Form == nil || page.Error == "" || page.Notice != "" || page.Reveal != nil {
t.Fatalf("uncertain creation state: %+v", page)
}
proxy.mu.Lock()
requests := slices.Clone(proxy.requests[before:])
proxy.mu.Unlock()
if !slices.Equal(requests, []string{"POST " + path}) {
t.Fatalf("creation replayed, configured a signature, or refreshed: %v", requests)
}
var original struct {
Result struct{ Hook daemon.Webhook } `json:"result"`
}
if err := json.Unmarshal(accepted, &original); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, err := daemon.RunWebhook(context.Background(), conn(t), id, daemon.WebhookRequest{Action: daemon.WebhookDelete, HookID: original.Result.Hook.ID})
if err != nil {
t.Error(err)
}
})
listed, err := daemon.RunWebhook(t.Context(), conn(t), id, daemon.WebhookRequest{Action: daemon.WebhookList})
if err != nil {
t.Fatal(err)
}
count := 0
for _, entry := range listed.Hooks {
if entry.Hook.Session == id && entry.Hook.Name == "missing-secret" {
count++
if entry.Hook.Header != "x-albedo-signature" {
t.Fatal("signature step ran without the generated secret")
}
}
}
if count != 1 {
t.Fatalf("persisted %d hooks, want one", count)
}
}
func TestUnknownReceiptReplaysOriginalCreationOnce(t *testing.T) {
providerRoute(t, echoReply)
workspace := t.TempDir()
proxy := cutAcknowledgement(t, "/sessions", hideReceiptAcknowledgement)
handle, err := daemon.NewCreation(daemon.CreateSessionRequest{Workspace: workspace})
if err != nil {
t.Fatal(err)
}
created, err := daemon.CreateSessionOperation(t.Context(), proxy.connection, handle)
if err != nil {
t.Fatal(err)
}
proxy.mu.Lock()
responses := slices.Clone(proxy.responses)
proxy.mu.Unlock()
if len(responses) != 2 || !bytes.Equal(responses[0], responses[1]) {
t.Fatalf("same intent replay produced different receipts: %q", responses)
}
count := 0
for _, session := range daemonSessions(t) {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("replay created %d sessions", count)
}
resolved, err := daemon.ResolveCreation(t.Context(), proxy.connection, handle)
if err != nil || resolved.ID != created.ID {
t.Fatalf("original creation: %+v %v", resolved, err)
}
}
func TestQueuedPromptDeadlineCancelsOnlyItsOperation(t *testing.T) {
entered, release := make(chan struct{}), make(chan struct{})
released := false
defer func() {
if !released {
close(release)
}
}()
profile := providerRoute(t, func(request map[string]any) string {
if lastUserText(request) == "active" {
close(entered)
<-release
}
return echoReply(request)
})
id := newSession(t, t.TempDir())
if _, err := daemon.NewChatClient(conn(t), id).Send(t.Context(), "active", nil); err != nil {
t.Fatal(err)
}
select {
case <-entered:
case <-time.After(10 * time.Second):
t.Fatal("active turn never started")
}
connection := daemon.NewConnection(conn(t).Snapshot(), nil)
counter := &submissionCounterTransport{next: connection.HTTPClient().Transport, accepted: make(chan struct{})}
connection.HTTPClient().Transport = counter
t.Cleanup(connection.HTTPClient().CloseIdleConnections)
service := app.Service{Connect: func(context.Context) (*daemon.Connection, error) { return connection, nil }}
finished := make(chan error, 1)
go func() {
_, err := service.RunPrompt(t.Context(), app.PromptOptions{SessionID: id, Prompt: "waiting", Timeout: 2 * time.Second})
finished <- err
}()
select {
case <-counter.accepted:
case err := <-finished:
t.Fatalf("prompt ended before durable acceptance: %v", err)
case <-time.After(5 * time.Second):
t.Fatal("queued prompt was not admitted")
}
err := <-finished
uncertain, ok := errors.AsType[*daemon.UncertainOutcomeError](err)
if !ok || uncertain.Handle == nil || !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("deadline lost operation identity: %v", err)
}
receipt, err := daemon.ResolveOperation(t.Context(), connection, uncertain.Handle)
if err != nil || receipt.DeliveryStatus != "cancelled" {
t.Fatalf("queued receipt after deadline: %+v %v", receipt, err)
}
if counter.posts.Load() != 1 {
t.Fatal("query recovery submitted another intent")
}
close(release)
released = true
waitIdle(t, id, profile, 1)
receipt, err = daemon.ResolveOperation(t.Context(), connection, uncertain.Handle)
if err != nil || receipt.DeliveryStatus != "cancelled" {
t.Fatalf("cancelled receipt after unrelated turn: %+v %v", receipt, err)
}
if counter.posts.Load() != 1 {
t.Fatal("query recovery submitted another intent")
}
history, err := daemon.NewChatClient(conn(t), id).History(t.Context(), 0, 120)
if err != nil {
t.Fatal(err)
}
matched := 0
for _, event := range history.Events {
if event.Type == daemon.EventUser && event.OperationID == uncertain.Handle.ID() {
matched++
}
}
if matched != 0 {
t.Fatalf("cancelled deadline operation delivered %d times", matched)
}
}
// Count the real client's submissions while leaving its SSE stream unbuffered.
type submissionCounterTransport struct {
next http.RoundTripper
posts atomic.Int32
accepted chan struct{}
acceptOnce sync.Once
}
func (transport *submissionCounterTransport) RoundTrip(request *http.Request) (*http.Response, error) {
if request.Method == http.MethodPost && strings.HasSuffix(request.URL.Path, "/events") {
transport.posts.Add(1)
}
response, err := transport.next.RoundTrip(request)
if err == nil && request.Method == http.MethodPost && strings.HasSuffix(request.URL.Path, "/events") && response.StatusCode == http.StatusAccepted {
transport.acceptOnce.Do(func() { close(transport.accepted) })
}
return response, err
}
func (transport *submissionCounterTransport) CloseIdleConnections() {
if closer, ok := transport.next.(interface{ CloseIdleConnections() }); ok {
closer.CloseIdleConnections()
}
}
func TestTUICreationRecoveryRetainsSelectedSession(t *testing.T) {
profile := providerRoute(t, echoReply)
initial := daemonSession(t, newFaultSession(t, profile))
other := daemonSession(t, newFaultSession(t, profile))
workspace := t.TempDir()
proxy := cutAcknowledgement(t, "/sessions", unresolvedAcknowledgement)
driver := driveTUIWithConnection(t, &initial, proxy.connection)
t.Cleanup(func() { driver.App.Chat.Close() })
uncertain := driver.Send(tui.FolderNewSessionMsg{Workspace: workspace})
driver.Update(tui.FolderOpenSessionMsg{Session: other})
recovery := driver.Update(uncertain)
if recovery == nil {
t.Fatal("navigation discarded uncertain creation recovery")
}
proxy.allowReceipt.Store(true)
for _, recoveryRequest := range driver.results(recovery) {
driver.Update(driver.Send(recoveryRequest))
}
if driver.App.ActiveSession == nil || driver.App.ActiveSession.ID != other.ID {
t.Fatal("recovering older creation replaced the user's selected session")
}
count := 0
for _, session := range driver.App.Sessions {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("navigation recovered %d created sessions", count)
}
proxy.mu.Lock()
responses := slices.Clone(proxy.responses)
proxy.mu.Unlock()
if len(responses) != 2 || !bytes.Equal(responses[0], responses[1]) {
t.Fatal("creation recovery submitted another intent")
}
count = 0
for _, session := range daemonSessions(t) {
if session.Workspace == workspace {
count++
}
}
if count != 1 {
t.Fatalf("creation uncertainty produced %d durable sessions", count)
}
}
// Receipt fault tests also navigate sessions, whose event streams must remain live.
type flushingResponseWriter struct{ http.ResponseWriter }
func (writer flushingResponseWriter) Write(data []byte) (int, error) {
count, err := writer.ResponseWriter.Write(data)
if err == nil {
err = http.NewResponseController(writer.ResponseWriter).Flush()
}
return count, err
}