tutti/packages/agent/daemon/runtime/codex_appserver_adapter_close_test.go

package agentruntime

import (
	"context"
	"testing"
	"time"
)

func TestCodexAppServerCloseQuiescesActiveTurnBeforeSharedDetach(t *testing.T) {
	adapter, transport, session := startedAppServerAdapter(t)
	transport.server.holdTurn = true
	parentAppSession := adapter.getSession(session.AgentSessionID)
	adapter.mu.Lock()
	adapter.sessions["side-shared"] = &codexAppServerSession{
		client:   parentAppSession.client,
		threadID: "codex-thread-side",
	}
	adapter.mu.Unlock()

	execDone := make(chan struct{}, 1)
	go func() {
		_, _ = adapter.Exec(context.Background(), session, []PromptContentBlock{{
			Type: "text", Text: "long task",
		}}, "", "turn-local-close", nil, nil)
		execDone <- struct{}{}
	}()
	waitForCondition(t, func() bool {
		return adapter.sessionActiveTurnID(session.AgentSessionID) == "turn-1"
	})

	if err := adapter.QuiesceForClose(context.Background(), session); err != nil {
		t.Fatalf("QuiesceForClose: %v", err)
	}
	if err := adapter.Close(context.Background(), session); err != nil {
		t.Fatalf("Close: %v", err)
	}
	select {
	case <-execDone:
	case <-time.After(5 * time.Second):
		t.Fatal("active Turn survived shared session close")
	}
	interrupt := appServerRequestParams(t, transport.conn, appServerMethodTurnInterrupt)
	if asString(interrupt["threadId"]) != "codex-thread-1" {
		t.Fatalf("turn/interrupt params = %#v", interrupt)
	}
	unsubscribe := appServerRequestParams(t, transport.conn, appServerMethodThreadUnsubscribe)
	if asString(unsubscribe["threadId"]) != "codex-thread-1" {
		t.Fatalf("thread/unsubscribe params = %#v", unsubscribe)
	}
	if adapter.getSession("side-shared") == nil {
		t.Fatal("closing parent removed shared Side session")
	}
	select {
	case <-transport.conn.closed:
		t.Fatal("closing parent terminated shared client")
	default:
	}
}

func TestCodexAppServerCloseWaitsForQueuedInterruptBeforeSharedDetach(t *testing.T) {
	adapter, transport, session := startedAppServerAdapter(t)
	transport.server.holdTurn = true
	transport.server.turnStartEntered = make(chan struct{})
	transport.server.turnStartRelease = make(chan struct{})
	parentAppSession := adapter.getSession(session.AgentSessionID)
	adapter.mu.Lock()
	adapter.sessions["side-shared-pending-start"] = &codexAppServerSession{
		client:   parentAppSession.client,
		threadID: "codex-thread-side",
	}
	adapter.mu.Unlock()

	execDone := make(chan struct{}, 1)
	go func() {
		_, _ = adapter.Exec(context.Background(), session, []PromptContentBlock{{
			Type: "text", Text: "long task",
		}}, "", "turn-local-close-pending-start", nil, nil)
		execDone <- struct{}{}
	}()
	select {
	case <-transport.server.turnStartEntered:
	case <-time.After(5 * time.Second):
		t.Fatal("turn/start was not sent")
	}
	waitForCondition(t, func() bool {
		return adapter.sessionActiveTurn(session.AgentSessionID) != nil &&
			adapter.sessionActiveTurnID(session.AgentSessionID) == ""
	})

	quiesced := make(chan error, 1)
	go func() {
		quiesced <- adapter.QuiesceForClose(context.Background(), session)
	}()
	select {
	case err := <-quiesced:
		t.Fatalf("QuiesceForClose returned before queued interrupt settled: %v", err)
	case <-time.After(50 * time.Millisecond):
	}
	if adapter.getSession(session.AgentSessionID) == nil {
		t.Fatal("source session detached before queued interrupt settled")
	}

	close(transport.server.turnStartRelease)
	select {
	case err := <-quiesced:
		if err != nil {
			t.Fatalf("QuiesceForClose: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("QuiesceForClose did not wait through queued interrupt")
	}
	if err := adapter.Close(context.Background(), session); err != nil {
		t.Fatalf("Close: %v", err)
	}
	select {
	case <-execDone:
	case <-time.After(5 * time.Second):
		t.Fatal("active Turn survived pending-start shared session close")
	}
	interrupt := appServerRequestParams(t, transport.conn, appServerMethodTurnInterrupt)
	if asString(interrupt["threadId"]) != "codex-thread-1" ||
		asString(interrupt["turnId"]) != "turn-1" {
		t.Fatalf("turn/interrupt params = %#v", interrupt)
	}
	if adapter.getSession("side-shared-pending-start") == nil {
		t.Fatal("closing parent removed shared Side session")
	}
}

func TestCodexAppServerCloseForceStopsTurnWhoseStartNeverAcknowledges(t *testing.T) {
	adapter, transport, session := startedAppServerAdapter(t)
	adapter.cancelGraceWindow = 20 * time.Millisecond
	transport.server.hangTurnStart = true
	transport.server.turnStartEntered = make(chan struct{})
	parentAppSession := adapter.getSession(session.AgentSessionID)
	adapter.mu.Lock()
	adapter.sessions["side-shared-hung-start"] = &codexAppServerSession{
		client:   parentAppSession.client,
		threadID: "codex-thread-side",
	}
	adapter.mu.Unlock()

	execDone := make(chan struct{}, 1)
	go func() {
		_, _ = adapter.Exec(context.Background(), session, []PromptContentBlock{{
			Type: "text", Text: "long task",
		}}, "", "turn-local-close-hung-start", nil, nil)
		execDone <- struct{}{}
	}()
	select {
	case <-transport.server.turnStartEntered:
	case <-time.After(5 * time.Second):
		t.Fatal("turn/start was not sent")
	}
	waitForCondition(t, func() bool {
		return adapter.sessionActiveTurn(session.AgentSessionID) != nil &&
			adapter.sessionActiveTurnID(session.AgentSessionID) == ""
	})

	if err := adapter.QuiesceForClose(context.Background(), session); err != nil {
		t.Fatalf("QuiesceForClose: %v", err)
	}
	select {
	case <-execDone:
	case <-time.After(5 * time.Second):
		t.Fatal("hung turn/start survived close quiesce")
	}
	select {
	case <-transport.conn.closed:
	default:
		t.Fatal("shared transport remained alive with an uninterruptible provider Turn")
	}
}