//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, 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}, nil, // mcpTokens — not exercised by integration tests ) 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