agentic-coding-workflow/internal/app/acp_integration_test.go

//go:build integration

// acp_integration_test.go — through-stack integration matrix for the ACP module
// (spec §11.3, 14 scenarios).
//
// SCOPE: This file lays out the matrix of 14 scenarios as a scaffold + sub-tests.
// Most scenarios are already covered at lower test layers; through-stack coverage
// here ensures the wiring (HTTP → SessionService → Supervisor → Relay → fake-agent)
// composes correctly under real testcontainers/Postgres + a real fake-agent
// subprocess + a real chi router.
//
// Coverage map (lower-layer test ↔ this file):
//
//	Scenario                       Lower-layer coverage              This file
//	-----------------------------  --------------------------------  -----------------
//	CreateSessionSuccess           E3 TestSupervisor_HappyPath       ✓ scaffold + test
//	WSStreamLiveEvents             G4 TestHandler_Session_E2E        deferred (G4 ≈)
//	PromptAndCancel                D8 TestRelay_Call_RoundTrip*      deferred
//	SpawnFailedBadBinary           E3 TestSupervisor_HangAgent       deferred
//	HandshakeTimeout               E3 TestSupervisor_HangAgent       deferred
//	ChildCrashed                   E3 TestSupervisor_AgentCrashes    deferred
//	StartupReaper                  D1 ResetStuckSessionsOnRestart    deferred
//	WSReconnectWithSince           D8 TestRelay_FanoutToSubscribers  deferred
//	FsOutsideWorktreeRejected      D4 fs/read_text_file unit         deferred
//	PerUserQuotaExceeded           F3 SessionService.Create unit     ✓ scaffold + test
//	BranchConflict                 F3 SessionService.Create unit     ✓ scaffold + test
//	MainWorktreeMutex              D1 AcquireMainWorktree unit       deferred
//	DeleteAgentKindInUse           A11 repo TestPgRepo_DeleteInUse   deferred
//	DisableAgentKindOK             A* AgentKindService unit          deferred
//
// DEFERRED scenarios remain as t.Skip stubs with the lower-layer test referenced.
// Future work: fill them in using the same suite + helpers below.
//
// Build/run:
//   go test -count=1 -tags=integration ./internal/app/... -run TestACPIntegration

package app_test

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	tcpg "github.com/testcontainers/testcontainers-go/modules/postgres"

	"github.com/yan1h/agent-coding-workflow/internal/acp"
	"github.com/yan1h/agent-coding-workflow/internal/acp/handlers"
	"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/db"
	"github.com/yan1h/agent-coding-workflow/internal/infra/logger"
	"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
	"github.com/yan1h/agent-coding-workflow/internal/project"
	"github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware"
	"github.com/yan1h/agent-coding-workflow/internal/user"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

func TestACPIntegration_Matrix(t *testing.T) {
	if testing.Short() {
		t.Skip("integration; needs docker + go build")
	}
	suite := newACPIntegrationSuite(t)
	t.Cleanup(suite.cleanup)

	// 14 scenarios per spec §11.3. Implemented = full through-stack assertion.
	// Deferred = covered at lower test layers; sub-test t.Skip's with the reference.
	t.Run("CreateSessionSuccess", suite.testCreateSessionSuccess)
	t.Run("WSStreamLiveEvents", deferredAt("G4 TestHandler_Session_CreateAndWS_E2E"))
	t.Run("PromptAndCancel", deferredAt("D8 TestRelay_*"))
	t.Run("SpawnFailedBadBinary", deferredAt("E3 TestSupervisor_HangAgent"))
	t.Run("HandshakeTimeout", deferredAt("E3 TestSupervisor_HangAgent_HandshakeTimeout"))
	t.Run("ChildCrashed", deferredAt("E3 TestSupervisor_AgentCrashes"))
	t.Run("StartupReaper", deferredAt("D1 unit + F4 reaper wiring inspection"))
	t.Run("WSReconnectWithSince", deferredAt("D8 TestRelay_FanoutToSubscribers"))
	t.Run("FsOutsideWorktreeRejected", deferredAt("D4 fs/read_text_file unit"))
	t.Run("PerUserQuotaExceeded", suite.testPerUserQuotaExceeded)
	t.Run("BranchConflict", suite.testBranchConflict)
	t.Run("MainWorktreeMutex", deferredAt("D1 AcquireMainWorktree unit"))
	t.Run("DeleteAgentKindInUse", deferredAt("A11 TestPgRepo_AgentKind_DeleteInUse"))
	t.Run("DisableAgentKindOK", deferredAt("A* AgentKindService unit"))
}

// deferredAt returns a sub-test that t.Skip's with a pointer to the lower-layer
// test that already covers the scenario.
func deferredAt(ref string) func(t *testing.T) {
	return func(t *testing.T) {
		t.Skipf("covered at lower layer: %s", ref)
	}
}

// acpIntegrationSuite holds shared state across the matrix sub-tests.
type acpIntegrationSuite struct {
	srv        *httptest.Server
	pool       *pgxpool.Pool
	adminToken string
	userToken  string
	wsID       uuid.UUID
	fakeBin    string
	cleanup    func()
}

func newACPIntegrationSuite(t *testing.T) *acpIntegrationSuite {
	t.Helper()
	if _, err := exec.LookPath("git"); err != nil {
		t.Skip("git not available")
	}

	ctx := context.Background()

	container, err := tcpg.Run(ctx, "postgres:16-alpine",
		tcpg.WithDatabase("acw"),
		tcpg.WithUsername("acw"),
		tcpg.WithPassword("acw"),
	)
	require.NoError(t, err)
	t.Cleanup(func() { _ = container.Terminate(ctx) })

	dsn, err := container.ConnectionString(ctx, "sslmode=disable")
	require.NoError(t, err)
	require.NoError(t, db.Migrate(ctx, dsn, "file://../../migrations"))

	pool, err := db.NewPool(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(pool.Close)

	log := logger.New(logger.Options{Format: logger.FormatJSON, Level: logger.LevelInfo})

	auditRec := audit.NewPostgresRecorder(pool)
	encKey := make([]byte, 32)
	for i := range encKey {
		encKey[i] = byte(i + 1)
	}
	enc, err := crypto.NewEncryptor(encKey)
	require.NoError(t, err)

	userRepo := user.NewPostgresRepository(pool)
	userSvc := user.NewService(userRepo, auditRec)

	notifyRepo := notify.NewPostgresRepository(pool)
	disp := notify.NewDispatcher(notifyRepo, log)

	adminUser, err := userSvc.Bootstrap(ctx, "admin@local", "password123")
	require.NoError(t, err)
	require.NotNil(t, adminUser)

	// Build the ACP supervisor + service against this DB.
	acpRepo := acp.NewPostgresRepository(pool)

	akSvc := acp.NewAgentKindService(acpRepo, enc, auditRec)

	supCfg := acp.SupervisorConfig{
		SpawnTimeout:       5 * time.Second,
		KillGrace:          1 * time.Second,
		StderrBufferLines:  50,
		StderrTailBytes:    1000,
		EventMaxPayload:    65536,
		EventTruncateField: 4096,
		ShutdownGrace:      5 * time.Second,
	}
	sup := acp.NewSupervisor(acpRepo, auditRec, disp, nil, nil, enc, supCfg, log)

	// Minimal stubs for SessionService dependencies — sufficient for branch=main
	// path (no wtSvc / wsRepo calls) and tests that exercise quota/conflict logic.
	wsStub := &acpStubWsSvc{}
	paStub := &acpStubProjectAccess{}

	sessSvc := acp.NewSessionService(
		acpRepo, wsStub, nil, nil, paStub, sup, auditRec,
		acp.SessionServiceConfig{MaxActiveGlobal: 20, MaxActivePerUser: 3},
	)

	r := chi.NewRouter()
	r.Use(middleware.RequestID)
	user.NewHandler(userSvc).Mount(r)
	acp.NewHandler(
		akSvc, sessSvc, sup, acpRepo,
		userSvc,
		acpUserAdminAdapter{svc: userSvc},
		enc,
		acp.WSConfig{PingInterval: 30 * time.Second, PongTimeout: 10 * time.Second, SendBuffer: 64},
	).Mount(r)

	srv := httptest.NewServer(r)
	t.Cleanup(srv.Close)

	// Login admin → adminToken
	adminToken := login(t, srv.URL, "admin@local", "password123")

	// Provision a regular user via direct SQL + Login (user.Service has no public
	// Create method; Bootstrap is idempotent admin-only). Hash matches user.NewService's
	// expectation: bcrypt of "password123" produced by user package's hashing helper.
	regularUserID := uuid.New()
	regularEmail := "regular@local"
	// Use Bootstrap on a fresh email to get a properly hashed user, then flip is_admin off.
	regularUser, err := userSvc.Bootstrap(ctx, regularEmail, "password123")
	require.NoError(t, err)
	require.NotNil(t, regularUser)
	regularUserID = regularUser.ID
	_, err = pool.Exec(ctx, `UPDATE users SET is_admin = false WHERE id = $1`, regularUserID)
	require.NoError(t, err)
	userToken := login(t, srv.URL, regularEmail, "password123")

	// Project + workspace rows. Direct SQL to avoid needing the full project handler
	// here — the ACP path uses workspace.WorkspaceService.Get (stubbed) so the rows
	// only need to exist for the ProjectAccess lookups SessionService performs.
	pid := uuid.New()
	wsID := uuid.New()
	cwd := t.TempDir()
	_, err = pool.Exec(ctx, `INSERT INTO projects (id, slug, name, owner_id) VALUES ($1, $2, 'p', $3)`,
		pid, "p-acp-int", regularUser.ID)
	require.NoError(t, err)
	_, err = pool.Exec(ctx, `INSERT INTO workspaces
		(id, project_id, slug, name, git_remote_url, main_path, default_branch)
		VALUES ($1, $2, $3, 'w', 'git@x:y.git', $4, 'main')`,
		wsID, pid, "ws-acp-int", cwd)
	require.NoError(t, err)

	// Acquire the workspace row for the stub to return.
	wsRepoLocal := workspace.NewPostgresRepository(pool)
	wsRow, err := wsRepoLocal.GetWorkspaceByID(ctx, wsID)
	require.NoError(t, err)
	wsStub.ws = wsRow
	paStub.pid = pid

	suite := &acpIntegrationSuite{
		srv:        srv,
		pool:       pool,
		adminToken: adminToken,
		userToken:  userToken,
		wsID:       wsID,
		fakeBin:    buildFakeAgent(t),
	}
	suite.cleanup = func() {
		sup.ShutdownAll(context.Background())
	}
	return suite
}

// === Implemented sub-tests ===

func (s *acpIntegrationSuite) testCreateSessionSuccess(t *testing.T) {
	akID := s.createAgentKind(t, "fake-success", s.fakeBin, nil)

	body, _ := json.Marshal(map[string]any{
		"workspace_id": s.wsID.String(), "agent_kind_id": akID, "branch": "main",
	})
	resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body)
	require.Equal(t, http.StatusCreated, resp.StatusCode)

	var dto struct {
		ID     string `json:"id"`
		Status string `json:"status"`
	}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto))
	_ = resp.Body.Close()
	assert.Equal(t, "starting", dto.Status)

	s.waitFor(t, 10*time.Second, func() bool {
		return s.getSession(t, dto.ID).Status == "running"
	})
}

func (s *acpIntegrationSuite) testPerUserQuotaExceeded(t *testing.T) {
	akID := s.createAgentKind(t, "fake-quota", s.fakeBin, nil)

	// MaxActivePerUser=3 (set in suite). Create 3 sessions on distinct branches.
	for i := 0; i < 3; i++ {
		body, _ := json.Marshal(map[string]any{
			"workspace_id":  s.wsID.String(),
			"agent_kind_id": akID,
			"branch":        "feat-q-" + strconv.Itoa(i),
		})
		resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body)
		require.Equal(t, http.StatusCreated, resp.StatusCode, "creating session %d", i)
		_ = resp.Body.Close()
	}

	// 4th must 429.
	body, _ := json.Marshal(map[string]any{
		"workspace_id":  s.wsID.String(),
		"agent_kind_id": akID,
		"branch":        "feat-q-overflow",
	})
	resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, body)
	defer resp.Body.Close()
	assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode)
}

func (s *acpIntegrationSuite) testBranchConflict(t *testing.T) {
	akID := s.createAgentKind(t, "fake-bc", s.fakeBin, nil)

	first, _ := json.Marshal(map[string]any{
		"workspace_id":  s.wsID.String(),
		"agent_kind_id": akID,
		"branch":        "main",
	})
	resp := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, first)
	require.Equal(t, http.StatusCreated, resp.StatusCode)
	_ = resp.Body.Close()

	// Second request on same main branch: main worktree mutex blocks → 409.
	resp2 := s.do(t, http.MethodPost, "/api/v1/acp/sessions", s.userToken, first)
	defer resp2.Body.Close()
	assert.Equal(t, http.StatusConflict, resp2.StatusCode)
}

// === Helper methods ===

func (s *acpIntegrationSuite) do(t *testing.T, method, path, token string, body []byte) *http.Response {
	t.Helper()
	var rd io.Reader
	if body != nil {
		rd = bytes.NewReader(body)
	}
	req, _ := http.NewRequest(method, s.srv.URL+path, rd)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+token)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	return resp
}

func (s *acpIntegrationSuite) createAgentKind(t *testing.T, name, bin string, env map[string]string) string {
	t.Helper()
	body := map[string]any{
		"name": name, "display_name": name, "binary_path": bin,
		"args": []string{}, "env": env, "enabled": true,
	}
	bjson, _ := json.Marshal(body)
	resp := s.do(t, http.MethodPost, "/api/v1/acp/agent-kinds", s.adminToken, bjson)
	defer resp.Body.Close()
	require.Equal(t, http.StatusCreated, resp.StatusCode)
	var dto struct {
		ID string `json:"id"`
	}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto))
	return dto.ID
}

type acpSessionDTO struct {
	ID     string `json:"id"`
	Status string `json:"status"`
}

func (s *acpIntegrationSuite) getSession(t *testing.T, sid string) acpSessionDTO {
	t.Helper()
	resp := s.do(t, http.MethodGet, "/api/v1/acp/sessions/"+sid, s.userToken, nil)
	defer resp.Body.Close()
	require.Equal(t, http.StatusOK, resp.StatusCode)
	var dto acpSessionDTO
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&dto))
	return dto
}

func (s *acpIntegrationSuite) waitFor(t *testing.T, d time.Duration, pred func() bool) {
	t.Helper()
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		if pred() {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatal("waitFor: timeout")
}

// === Stubs / adapters ===

type acpUserAdminAdapter struct{ svc user.Service }

func (a acpUserAdminAdapter) IsAdmin(ctx context.Context, id uuid.UUID) (bool, error) {
	u, err := a.svc.Get(ctx, id)
	if err != nil {
		return false, err
	}
	return u.IsAdmin, nil
}

// acpStubWsSvc is the minimal workspace.WorkspaceService stub needed for the
// branch=main happy path (only Get is called).
type acpStubWsSvc struct {
	ws *workspace.Workspace
}

func (s *acpStubWsSvc) Get(_ context.Context, _ workspace.Caller, _ uuid.UUID) (*workspace.Workspace, error) {
	return s.ws, nil
}

func (s *acpStubWsSvc) Create(context.Context, workspace.Caller, string, workspace.CreateWorkspaceInput) (*workspace.Workspace, error) {
	return nil, fmt.Errorf("acpStubWsSvc.Create unused")
}

func (s *acpStubWsSvc) List(context.Context, workspace.Caller, string) ([]*workspace.Workspace, error) {
	return nil, fmt.Errorf("acpStubWsSvc.List unused")
}

func (s *acpStubWsSvc) Update(context.Context, workspace.Caller, uuid.UUID, workspace.UpdateWorkspaceInput) (*workspace.Workspace, error) {
	return nil, fmt.Errorf("acpStubWsSvc.Update unused")
}

func (s *acpStubWsSvc) Delete(context.Context, workspace.Caller, uuid.UUID) error {
	return fmt.Errorf("acpStubWsSvc.Delete unused")
}

func (s *acpStubWsSvc) Sync(context.Context, workspace.Caller, uuid.UUID) (*workspace.Workspace, error) {
	return nil, fmt.Errorf("acpStubWsSvc.Sync unused")
}

func (s *acpStubWsSvc) UpsertCredential(context.Context, workspace.Caller, uuid.UUID, workspace.UpsertCredentialInput) (*workspace.Credential, error) {
	return nil, fmt.Errorf("acpStubWsSvc.UpsertCredential unused")
}

func (s *acpStubWsSvc) GetCredentialMeta(context.Context, workspace.Caller, uuid.UUID) (*workspace.Credential, error) {
	return nil, fmt.Errorf("acpStubWsSvc.GetCredentialMeta unused")
}

// acpStubProjectAccess satisfies acp.ProjectAccess; only GetProjectByWorkspace is
// exercised by the branch=main path.
type acpStubProjectAccess struct {
	pid uuid.UUID
}

func (s *acpStubProjectAccess) GetProjectByWorkspace(_ context.Context, _ uuid.UUID) (*project.Project, error) {
	return &project.Project{ID: s.pid, Slug: "p-acp-int"}, nil
}

func (s *acpStubProjectAccess) GetIssueByNumber(context.Context, uuid.UUID, int) (*project.Issue, error) {
	return nil, fmt.Errorf("acpStubProjectAccess.GetIssueByNumber unused")
}

func (s *acpStubProjectAccess) GetRequirementByNumber(context.Context, uuid.UUID, int) (*project.Requirement, error) {
	return nil, fmt.Errorf("acpStubProjectAccess.GetRequirementByNumber unused")
}

// === Helpers ===

// login posts /api/v1/auth/login and returns the bearer token.
func login(t *testing.T, baseURL, email, password string) string {
	t.Helper()
	body, _ := json.Marshal(map[string]string{"email": email, "password": password})
	resp, err := http.Post(baseURL+"/api/v1/auth/login", "application/json", bytes.NewReader(body))
	require.NoError(t, err)
	defer resp.Body.Close()
	require.Equal(t, http.StatusOK, resp.StatusCode)
	var lr struct {
		Token string `json:"token"`
	}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&lr))
	require.NotEmpty(t, lr.Token)
	return lr.Token
}

// buildFakeAgent compiles cmd/fake-acp-agent into t.TempDir and returns the path.
func buildFakeAgent(t *testing.T) string {
	t.Helper()
	bin := filepath.Join(t.TempDir(), "fake-acp-agent")
	if runtime.GOOS == "windows" {
		bin += ".exe"
	}
	wd, _ := os.Getwd()
	for {
		if _, err := os.Stat(filepath.Join(wd, "go.mod")); err == nil {
			break
		}
		parent := filepath.Dir(wd)
		if parent == wd {
			t.Fatal("repo root not found")
		}
		wd = parent
	}
	cmd := exec.Command("go", "build", "-o", bin, "./cmd/fake-acp-agent")
	cmd.Dir = wd
	out, err := cmd.CombinedOutput()
	require.NoError(t, err, "build fake-acp-agent: %s", out)
	return bin
}

// suppress unused import warnings while scaffold is partial:
var _ = handlers.NewFsHandler
var _ = strings.HasPrefix