agentic-coding-workflow/internal/acp/handler_e2e_test.go

//go:build integration

// handler_e2e_test.go drives the ACP handler through one full happy-path
// session lifecycle:
//
//  1. POST /api/v1/acp/sessions → 201
//  2. wait for supervisor handshake → SessionRunning
//  3. WS /api/v1/acp/sessions/{id}/ws → expect "state" event
//  4. DELETE /api/v1/acp/sessions/{id} → 204
//  5. wait for monitor → SessionExited
//
// Build-tagged integration: spins up testcontainer PG + go build fake-acp-agent,
// so it never runs under -short.
package acp_test

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	ws "github.com/coder/websocket"
	"github.com/coder/websocket/wsjson"

	"github.com/yan1h/agent-coding-workflow/internal/acp"
	"github.com/yan1h/agent-coding-workflow/internal/audit"
	"github.com/yan1h/agent-coding-workflow/internal/infra/crypto"
	"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
	"github.com/yan1h/agent-coding-workflow/internal/project"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

// stubWsService satisfies workspace.WorkspaceService for the e2e test.
// Only Get is exercised by SessionService.Create; other methods panic.
type stubWsService struct {
	wsRow *workspace.Workspace
}

func (s *stubWsService) Get(_ context.Context, _ workspace.Caller, _ uuid.UUID) (*workspace.Workspace, error) {
	return s.wsRow, nil
}
func (s *stubWsService) Create(context.Context, workspace.Caller, string, workspace.CreateWorkspaceInput) (*workspace.Workspace, error) {
	panic("stubWsService.Create unused")
}
func (s *stubWsService) List(context.Context, workspace.Caller, string) ([]*workspace.Workspace, error) {
	panic("stubWsService.List unused")
}
func (s *stubWsService) Update(context.Context, workspace.Caller, uuid.UUID, workspace.UpdateWorkspaceInput) (*workspace.Workspace, error) {
	panic("stubWsService.Update unused")
}
func (s *stubWsService) Delete(context.Context, workspace.Caller, uuid.UUID) error {
	panic("stubWsService.Delete unused")
}
func (s *stubWsService) Sync(context.Context, workspace.Caller, uuid.UUID) (*workspace.Workspace, error) {
	panic("stubWsService.Sync unused")
}
func (s *stubWsService) UpsertCredential(context.Context, workspace.Caller, uuid.UUID, workspace.UpsertCredentialInput) (*workspace.Credential, error) {
	panic("stubWsService.UpsertCredential unused")
}
func (s *stubWsService) GetCredentialMeta(context.Context, workspace.Caller, uuid.UUID) (*workspace.Credential, error) {
	panic("stubWsService.GetCredentialMeta unused")
}
func (s *stubWsService) ListBranches(context.Context, workspace.Caller, uuid.UUID) ([]string, error) {
	panic("stubWsService.ListBranches unused")
}

// stubProjectAccess satisfies acp.ProjectAccess for the e2e test.
// Only GetProjectByWorkspace is exercised; the rest panic.
type stubProjectAccess struct {
	pj *project.Project
}

func (s *stubProjectAccess) GetProjectByWorkspace(_ context.Context, _ uuid.UUID) (*project.Project, error) {
	return s.pj, nil
}
func (s *stubProjectAccess) GetIssueByNumber(context.Context, uuid.UUID, int) (*project.Issue, error) {
	panic("stubProjectAccess.GetIssueByNumber unused")
}
func (s *stubProjectAccess) GetRequirementByNumber(context.Context, uuid.UUID, int) (*project.Requirement, error) {
	panic("stubProjectAccess.GetRequirementByNumber unused")
}

// TestHandler_Session_CreateAndWS_E2E covers the happy path from POST /sessions
// through WS state event to DELETE — the full request → subprocess → relay →
// shutdown loop, exercising Handler + SessionService + Supervisor end-to-end.
func TestHandler_Session_CreateAndWS_E2E(t *testing.T) {
	if testing.Short() {
		t.Skip("e2e: spawns subprocess + testcontainer pg")
	}
	t.Parallel()
	ctx := context.Background()

	repo, pool := setupRepo(t)

	uid := mustInsertUser(t, ctx, pool, "wsuser-e2e@local", false)
	pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)

	// mustInsertProjectWorkspace hardcodes main_path='/tmp'. Override to a real
	// per-test directory so subprocess Spawn can chdir into it.
	cwd := t.TempDir()
	_, err := pool.Exec(ctx,
		`UPDATE workspaces SET main_path = $1 WHERE id = $2`, cwd, wsid)
	require.NoError(t, err)

	// migrations/0003 sets default_branch DEFAULT 'main', so workspaces row
	// inserted by the helper already has DefaultBranch='main' — no UPDATE needed.

	bin := fakeAgentBinary(t)
	enc := testEncryptor(t)

	kind, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
		ID: uuid.New(), Name: "ws-e2e-fake", DisplayName: "Fake",
		BinaryPath: bin, Args: []string{}, Enabled: true, CreatedBy: uid,
	})
	require.NoError(t, err)

	// Real workspace row → seeded into the wsService stub so SessionService.Get
	// returns the same ProjectID + DefaultBranch the DB has.
	wsRow, err := workspace.NewPostgresRepository(pool).GetWorkspaceByID(ctx, wsid)
	require.NoError(t, err)

	disp := notify.NewDispatcher(notifyRepoStub{}, slogDiscard())
	sup := acp.NewSupervisor(
		repo,
		audit.NewPostgresRecorder(pool),
		disp,
		nil, nil, // wtSvc/wsRepo: IsMainWorktree=true path → not called
		nil, // mcpTokens
		enc,
		acp.SupervisorConfig{
			SpawnTimeout:       5 * time.Second,
			KillGrace:          1 * time.Second,
			StderrBufferLines:  50,
			StderrTailBytes:    1000,
			EventMaxPayload:    65536,
			EventTruncateField: 4096,
			ShutdownGrace:      5 * time.Second,
		},
		slogDiscard(),
	)
	defer sup.ShutdownAll(context.Background())

	wsStub := &stubWsService{wsRow: wsRow}
	paStub := &stubProjectAccess{pj: &project.Project{ID: pid, Slug: "p-test"}}

	svc := acp.NewSessionService(
		repo, wsStub, nil, nil, paStub, nil, nil, sup,
		audit.NewPostgresRecorder(pool),
		acp.SessionServiceConfig{MaxActiveGlobal: 10, MaxActivePerUser: 5},
		nil, // mcpTokens — not exercised by handler e2e
	)

	permSvc := acp.NewPermissionService(repo, nil, 0, nil)
	sup.SetPermissionService(permSvc)
	permSvc.SetRelayLocator(sup)
	h := acp.NewHandler(
		nil, // ak svc — not exercised by sessions endpoints
		svc, sup, repo, permSvc,
		fakeResolver{uid: uid},
		fakeAdminLookup{admin: map[uuid.UUID]bool{uid: false}},
		enc,
		acp.WSConfig{
			PingInterval: 30 * time.Second,
			PongTimeout:  10 * time.Second,
			SendBuffer:   64,
		},
	)

	r := chi.NewRouter()
	h.Mount(r)
	srv := httptest.NewServer(r)
	defer srv.Close()

	// 1. POST /api/v1/acp/sessions
	branch := "main"
	body, err := json.Marshal(map[string]any{
		"workspace_id":  wsid.String(),
		"agent_kind_id": kind.ID.String(),
		"branch":        branch,
	})
	require.NoError(t, err)

	req, err := http.NewRequest("POST", srv.URL+"/api/v1/acp/sessions", bytes.NewReader(body))
	require.NoError(t, err)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer valid")

	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer resp.Body.Close()

	var created map[string]any
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&created))
	require.Equal(t, http.StatusCreated, resp.StatusCode, "body=%v", created)

	sidStr, _ := created["id"].(string)
	require.NotEmpty(t, sidStr)
	sid := uuid.MustParse(sidStr)

	// 2. Wait for handshake → SessionRunning (supervisor.Spawn runs in a goroutine
	// inside SessionService.Create, so the row starts as 'starting'). 10s budget
	// covers fake-agent stdin/stdout handshake on slow CI.
	deadline := time.After(10 * time.Second)
	for {
		s, _ := repo.GetSessionByID(ctx, sid)
		if s != nil && s.Status == acp.SessionRunning {
			break
		}
		select {
		case <-deadline:
			t.Fatalf("session never reached SessionRunning")
		case <-time.After(100 * time.Millisecond):
		}
	}

	// 3. WS connect — Authorization header is honored by the Auth middleware.
	wsURL := strings.Replace(srv.URL, "http", "ws", 1) +
		"/api/v1/acp/sessions/" + sid.String() + "/ws"
	wsCtx, wsCancel := context.WithTimeout(ctx, 5*time.Second)
	defer wsCancel()

	wsConn, _, err := ws.Dial(wsCtx, wsURL, &ws.DialOptions{
		HTTPHeader: http.Header{"Authorization": []string{"Bearer valid"}},
	})
	require.NoError(t, err)
	defer wsConn.Close(ws.StatusInternalError, "")

	var stateEv map[string]any
	require.NoError(t, wsjson.Read(wsCtx, wsConn, &stateEv))
	assert.Equal(t, "state", stateEv["kind"])
	assert.Equal(t, sid.String(), stateEv["session_id"])
	assert.Equal(t, "running", stateEv["status"])
	assert.Equal(t, "main", stateEv["branch"])

	// Close WS before DELETE so the subscriber drains and Kill doesn't race
	// the WS goroutine.
	_ = wsConn.Close(ws.StatusNormalClosure, "test done")

	// 4. DELETE /api/v1/acp/sessions/{id}
	delReq, err := http.NewRequest("DELETE",
		srv.URL+"/api/v1/acp/sessions/"+sid.String(), nil)
	require.NoError(t, err)
	delReq.Header.Set("Authorization", "Bearer valid")

	delResp, err := http.DefaultClient.Do(delReq)
	require.NoError(t, err)
	defer delResp.Body.Close()
	assert.Equal(t, http.StatusNoContent, delResp.StatusCode)

	// 5. Wait for terminal status. Kill blocks on monitor.done internally,
	// so by the time DELETE returns the row should already be terminal —
	// poll briefly to guard against scheduler latency.
	deadline2 := time.After(5 * time.Second)
	for {
		s, _ := repo.GetSessionByID(ctx, sid)
		if s != nil && (s.Status == acp.SessionExited || s.Status == acp.SessionCrashed) {
			return
		}
		select {
		case <-deadline2:
			s, _ := repo.GetSessionByID(ctx, sid)
			lastStatus := acp.SessionStatus("?")
			if s != nil {
				lastStatus = s.Status
			}
			t.Fatalf("session not terminated, last status=%s", lastStatus)
		case <-time.After(100 * time.Millisecond):
		}
	}
}

// TestSession_TurnCompletion_E2E drives a full session with an InitialPrompt
// against the fake-acp-agent (which emits session/update chunks then a
// session/prompt response with stopReason=end_turn). It asserts:
//   - an acp_turns row is created and completed with stop_reason=end_turn
//   - update_count == FAKE_ACP_PROMPT_UPDATES
//   - acp_sessions.last_stop_reason is set to end_turn
//   - a test TurnBus subscriber receives turn_completed + session_idle
func TestSession_TurnCompletion_E2E(t *testing.T) {
	if testing.Short() {
		t.Skip("e2e: spawns subprocess + testcontainer pg")
	}
	t.Parallel()
	ctx := context.Background()

	repo, pool := setupRepo(t)

	uid := mustInsertUser(t, ctx, pool, "turn-e2e@local", false)
	pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)
	cwd := t.TempDir()
	_, err := pool.Exec(ctx, `UPDATE workspaces SET main_path = $1 WHERE id = $2`, cwd, wsid)
	require.NoError(t, err)

	bin := fakeAgentBinary(t)
	enc := testEncryptor(t)

	const wantUpdates = 4
	kind, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
		ID: uuid.New(), Name: "turn-e2e-fake", DisplayName: "Fake",
		BinaryPath: bin, Args: []string{}, Enabled: true, CreatedBy: uid,
		// FAKE_ACP_PROMPT_UPDATES via encrypted env: emit N session/update before response.
		EncryptedEnv: mustEncryptEnv(t, enc, map[string]string{"FAKE_ACP_PROMPT_UPDATES": "4"}),
	})
	require.NoError(t, err)

	wsRow, err := workspace.NewPostgresRepository(pool).GetWorkspaceByID(ctx, wsid)
	require.NoError(t, err)

	disp := notify.NewDispatcher(notifyRepoStub{}, slogDiscard())
	sup := acp.NewSupervisor(
		repo, audit.NewPostgresRecorder(pool), disp,
		nil, nil, nil, enc,
		acp.SupervisorConfig{
			SpawnTimeout:       5 * time.Second,
			KillGrace:          1 * time.Second,
			StderrBufferLines:  50,
			StderrTailBytes:    1000,
			EventMaxPayload:    65536,
			EventTruncateField: 4096,
			ShutdownGrace:      5 * time.Second,
		},
		slogDiscard(),
	)
	defer sup.ShutdownAll(context.Background())

	// Test TurnBus subscriber records published events.
	bus := acp.NewTurnBus(slogDiscard())
	evCh := make(chan acp.TurnEvent, 8)
	bus.Subscribe("test", func(_ context.Context, ev acp.TurnEvent) { evCh <- ev })
	sup.SetTurnBus(bus)

	wsStub := &stubWsService{wsRow: wsRow}
	paStub := &stubProjectAccess{pj: &project.Project{ID: pid, Slug: "p-turn"}}
	svc := acp.NewSessionService(
		repo, wsStub, nil, nil, paStub, nil, nil, sup,
		audit.NewPostgresRecorder(pool),
		acp.SessionServiceConfig{MaxActiveGlobal: 10, MaxActivePerUser: 5},
		nil,
	)
	permSvc := acp.NewPermissionService(repo, nil, 0, nil)
	sup.SetPermissionService(permSvc)
	permSvc.SetRelayLocator(sup)

	prompt := "do the thing"
	branch := "main"
	sess, err := svc.Create(ctx, acp.Caller{UserID: uid}, acp.CreateSessionInput{
		WorkspaceID: wsid, AgentKindID: kind.ID, Branch: &branch, InitialPrompt: &prompt,
	})
	require.NoError(t, err)

	// Poll for the turn to complete.
	var completedTurn *acp.Turn
	deadline := time.After(15 * time.Second)
	for completedTurn == nil {
		ts, _ := repo.ListTurnsBySession(ctx, sess.ID)
		for _, tn := range ts {
			if tn.Status == acp.TurnCompleted {
				completedTurn = tn
			}
		}
		if completedTurn != nil {
			break
		}
		select {
		case <-deadline:
			t.Fatalf("turn never completed; turns=%+v", ts)
		case <-time.After(150 * time.Millisecond):
		}
	}

	require.NotNil(t, completedTurn.StopReason)
	assert.Equal(t, "end_turn", *completedTurn.StopReason)
	assert.Equal(t, wantUpdates, completedTurn.UpdateCount)

	got, _ := repo.GetSessionByID(ctx, sess.ID)
	require.NotNil(t, got.LastStopReason)
	assert.Equal(t, "end_turn", *got.LastStopReason)

	// Collect bus events: expect a turn_completed and a session_idle.
	var sawCompleted, sawIdle bool
	evDeadline := time.After(3 * time.Second)
	for !(sawCompleted && sawIdle) {
		select {
		case ev := <-evCh:
			switch ev.Kind {
			case acp.TurnCompletedEvent:
				sawCompleted = true
				assert.Equal(t, acp.StopEndTurn, ev.StopReason)
			case acp.SessionIdleEvent:
				sawIdle = true
			}
		case <-evDeadline:
			t.Fatalf("did not receive both turn_completed and session_idle (completed=%v idle=%v)", sawCompleted, sawIdle)
		}
	}
}

// mustEncryptEnv marshals env to JSON and encrypts it (mirrors acp.encryptEnv,
// which is unexported), so the supervisor can decrypt it back into the spawn env.
func mustEncryptEnv(t *testing.T, enc *crypto.Encryptor, env map[string]string) []byte {
	t.Helper()
	b, err := json.Marshal(env)
	require.NoError(t, err)
	out, err := enc.Encrypt(b)
	require.NoError(t, err)
	return out
}