test(acp): integration matrix scaffold + 3 implemented scenarios (CreateSuccess/Quota/BranchConflict)

internal/app/acp_integration_test.go

@@ -0,0 +1,498 @@
+//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