package acp_test import ( "context" "fmt" "sync" "testing" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/yan1h/agent-coding-workflow/internal/acp" "github.com/yan1h/agent-coding-workflow/internal/mcp" "github.com/yan1h/agent-coding-workflow/internal/project" "github.com/yan1h/agent-coding-workflow/internal/workspace" ) // fakeProjectAccess 满足 acp.ProjectAccess 接口;测试时按需返回。 type fakeProjectAccess struct { pj *project.Project issue *project.Issue req *project.Requirement err error } func (f fakeProjectAccess) GetProjectByWorkspace(_ context.Context, _ uuid.UUID) (*project.Project, error) { return f.pj, f.err } func (f fakeProjectAccess) GetIssueByNumber(_ context.Context, _ uuid.UUID, _ int) (*project.Issue, error) { if f.issue == nil { return nil, f.err } return f.issue, nil } func (f fakeProjectAccess) GetRequirementByNumber(_ context.Context, _ uuid.UUID, _ int) (*project.Requirement, error) { if f.req == nil { return nil, f.err } return f.req, nil } func TestResolveBranch_Explicit(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), DefaultBranch: "main"} br := "feat/x" got, isMain, err := acp.ResolveBranch(context.Background(), ws, uuid.New(), acp.CreateSessionInput{Branch: &br}, "demo", fakeProjectAccess{}) require.NoError(t, err) assert.Equal(t, "feat/x", got) assert.False(t, isMain) } func TestResolveBranch_ExplicitMain(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), DefaultBranch: "main"} br := "main" got, isMain, err := acp.ResolveBranch(context.Background(), ws, uuid.New(), acp.CreateSessionInput{Branch: &br}, "demo", fakeProjectAccess{}) require.NoError(t, err) assert.Equal(t, "main", got) assert.True(t, isMain) } func TestResolveBranch_FromIssue(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), ProjectID: uuid.New(), DefaultBranch: "main"} issueNum := 42 pa := fakeProjectAccess{issue: &project.Issue{Number: 42, WorkspaceID: nil}} got, isMain, err := acp.ResolveBranch(context.Background(), ws, uuid.New(), acp.CreateSessionInput{IssueNumber: &issueNum}, "demo", pa) require.NoError(t, err) assert.Equal(t, "issue/demo-42", got) assert.False(t, isMain) } func TestResolveBranch_IssueLinkedToOtherWorkspace(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), ProjectID: uuid.New(), DefaultBranch: "main"} other := uuid.New() pa := fakeProjectAccess{issue: &project.Issue{Number: 42, WorkspaceID: &other}} num := 42 _, _, err := acp.ResolveBranch(context.Background(), ws, uuid.New(), acp.CreateSessionInput{IssueNumber: &num}, "demo", pa) require.Error(t, err) } func TestResolveBranch_FromRequirement(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), ProjectID: uuid.New(), DefaultBranch: "main"} num := 7 pa := fakeProjectAccess{req: &project.Requirement{Number: 7, WorkspaceID: nil}} got, isMain, err := acp.ResolveBranch(context.Background(), ws, uuid.New(), acp.CreateSessionInput{RequirementNumber: &num}, "demo", pa) require.NoError(t, err) assert.Equal(t, "req/demo-7", got) assert.False(t, isMain) } func TestResolveBranch_AutoFallback(t *testing.T) { t.Parallel() ws := &workspace.Workspace{ID: uuid.New(), ProjectID: uuid.New(), DefaultBranch: "main"} sid := uuid.New() got, isMain, err := acp.ResolveBranch(context.Background(), ws, sid, acp.CreateSessionInput{}, "demo", fakeProjectAccess{}) require.NoError(t, err) assert.Contains(t, got, "acp-auto/") assert.False(t, isMain) } // ===== fakes for SessionService.Create tests ===== // fakeAcpRepo satisfies acp.Repository for Create-path unit tests. type fakeAcpRepo struct { mu sync.Mutex sessions []*acp.Session kinds []*acp.AgentKind inserted *acp.Session // txStaging holds sessions inserted during InTx; committed to sessions only // when fn returns nil (simulating real transaction semantics). txStaging []*acp.Session inTx bool // run-history dashboard: 记录入参 filter 供断言;返回预置结果。 rollupFilter acp.DashboardFilter byDayFilter acp.DashboardFilter rollupResult acp.SessionRollup byDayResult []acp.SessionDayRow timelineResult []acp.SessionTimelineBucket timelineSID uuid.UUID } func (r *fakeAcpRepo) InTx(_ context.Context, fn func(context.Context, pgx.Tx) error) error { r.mu.Lock() r.inTx = true r.txStaging = nil r.mu.Unlock() err := fn(context.Background(), nil) r.mu.Lock() defer r.mu.Unlock() r.inTx = false if err == nil { r.sessions = append(r.sessions, r.txStaging...) } r.txStaging = nil return err } func (r *fakeAcpRepo) WithTx(_ pgx.Tx) acp.Repository { return r } func (r *fakeAcpRepo) ListConfigFiles(context.Context, uuid.UUID) ([]*acp.ConfigFile, error) { return nil, nil } func (r *fakeAcpRepo) UpsertConfigFile(context.Context, *acp.ConfigFile) (*acp.ConfigFile, error) { panic("n/a") } func (r *fakeAcpRepo) DeleteConfigFile(context.Context, uuid.UUID, string) (bool, error) { panic("n/a") } func (r *fakeAcpRepo) InsertSession(_ context.Context, s *acp.Session) (*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() r.inserted = s if r.inTx { r.txStaging = append(r.txStaging, s) } else { r.sessions = append(r.sessions, s) } return s, nil } func (r *fakeAcpRepo) GetSessionByID(_ context.Context, id uuid.UUID) (*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() for _, s := range r.sessions { if s.ID == id { return s, nil } } return nil, fmt.Errorf("not found") } func (r *fakeAcpRepo) GetSessionByStepID(_ context.Context, stepID uuid.UUID) (*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() for _, s := range r.sessions { if s.OrchestratorStepID != nil && *s.OrchestratorStepID == stepID { return s, nil } } return nil, fmt.Errorf("not found") } func (r *fakeAcpRepo) GetCrashedSessionForResume(_ context.Context, stepID uuid.UUID) (*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() for _, s := range r.sessions { if s.OrchestratorStepID != nil && *s.OrchestratorStepID == stepID && (s.Status == acp.SessionCrashed || s.Status == acp.SessionExited) { return s, nil } } return nil, fmt.Errorf("not found") } func (r *fakeAcpRepo) ResetSessionForResume(_ context.Context, id uuid.UUID) error { r.mu.Lock() defer r.mu.Unlock() for _, s := range r.sessions { if s.ID == id { s.Status = acp.SessionStarting s.AgentSessionID = nil s.PID = nil s.ExitCode = nil s.LastError = nil s.EndedAt = nil return nil } } return fmt.Errorf("not found") } func (r *fakeAcpRepo) GetAgentKindByID(_ context.Context, id uuid.UUID) (*acp.AgentKind, error) { for _, k := range r.kinds { if k.ID == id { return k, nil } } return nil, fmt.Errorf("not found") } func (r *fakeAcpRepo) CountActiveSessions(_ context.Context) (int64, error) { r.mu.Lock() defer r.mu.Unlock() var n int64 for _, s := range r.sessions { if s.Status == acp.SessionStarting || s.Status == acp.SessionRunning { n++ } } return n, nil } func (r *fakeAcpRepo) CountActiveSessionsByUser(_ context.Context, _ uuid.UUID) (int64, error) { return 0, nil } func (r *fakeAcpRepo) AcquireMainWorktree(_ context.Context, _, _ uuid.UUID) (bool, error) { return true, nil } func (r *fakeAcpRepo) ReleaseMainWorktree(_ context.Context, _, _ uuid.UUID) (bool, error) { return true, nil } // Stub out remaining Repository methods (not used by Create path). func (r *fakeAcpRepo) CreateAgentKind(_ context.Context, k *acp.AgentKind) (*acp.AgentKind, error) { return k, nil } func (r *fakeAcpRepo) GetAgentKindByName(_ context.Context, _ string) (*acp.AgentKind, error) { return nil, fmt.Errorf("not found") } func (r *fakeAcpRepo) ListAgentKinds(_ context.Context) ([]*acp.AgentKind, error) { return nil, nil } func (r *fakeAcpRepo) ListEnabledAgentKinds(_ context.Context) ([]*acp.AgentKind, error) { return r.kinds, nil } func (r *fakeAcpRepo) UpdateAgentKind(_ context.Context, k *acp.AgentKind) (*acp.AgentKind, error) { return k, nil } func (r *fakeAcpRepo) DeleteAgentKind(_ context.Context, _ uuid.UUID) error { return nil } func (r *fakeAcpRepo) CountAgentKindUsage(_ context.Context, _ uuid.UUID) (int64, error) { return 0, nil } func (r *fakeAcpRepo) ListSessionsByUser(_ context.Context, userID uuid.UUID, reqID *uuid.UUID) ([]*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() var out []*acp.Session for _, s := range r.sessions { if s.UserID != userID { continue } if reqID != nil && (s.RequirementID == nil || *s.RequirementID != *reqID) { continue } out = append(out, s) } return out, nil } func (r *fakeAcpRepo) ListAllSessions(_ context.Context, reqID *uuid.UUID) ([]*acp.Session, error) { r.mu.Lock() defer r.mu.Unlock() var out []*acp.Session for _, s := range r.sessions { if reqID != nil && (s.RequirementID == nil || *s.RequirementID != *reqID) { continue } out = append(out, s) } return out, nil } func (r *fakeAcpRepo) UpdateSessionRunning(_ context.Context, _ uuid.UUID, _ string, _ int32) error { return nil } func (r *fakeAcpRepo) UpdateSessionSandboxMode(_ context.Context, _ uuid.UUID, _ string) error { return nil } func (r *fakeAcpRepo) UpdateSessionFinished(_ context.Context, _ uuid.UUID, _ acp.SessionStatus, _ *int32, _ *string) error { return nil } func (r *fakeAcpRepo) MarkSessionFailedIfActive(_ context.Context, id uuid.UUID, lastErr string) (bool, error) { r.mu.Lock() defer r.mu.Unlock() for _, s := range r.sessions { if s.ID == id && s.Status.IsActive() { s.Status = acp.SessionCrashed msg := lastErr s.LastError = &msg return true, nil } } return false, nil } func (r *fakeAcpRepo) ResetStuckSessionsOnRestart(_ context.Context) ([]*acp.Session, error) { return nil, nil } func (r *fakeAcpRepo) InsertEvent(_ context.Context, e *acp.Event) (*acp.Event, error) { return e, nil } func (r *fakeAcpRepo) ListEventsSince(_ context.Context, _ uuid.UUID, _ int64, _ int32) ([]*acp.Event, error) { return nil, nil } func (r *fakeAcpRepo) PurgeEventsBefore(_ context.Context, _ time.Time) (int64, error) { return 0, nil } func (r *fakeAcpRepo) InsertSessionUsage(_ context.Context, _ *acp.SessionUsageRecord) (bool, error) { return true, nil } func (r *fakeAcpRepo) AddSessionUsageTotals(_ context.Context, _ uuid.UUID, dp, dc, dt int64, dCost float64) (int64, int64, int64, float64, error) { return dp, dc, dt, dCost, nil } func (r *fakeAcpRepo) SumProjectCostUSD(_ context.Context, _ uuid.UUID, _ time.Time) (float64, error) { return 0, nil } func (r *fakeAcpRepo) MarkSessionTerminatedReason(_ context.Context, _ uuid.UUID, _ string) error { return nil } func (r *fakeAcpRepo) ListSessionsForReaper(_ context.Context, _, _ time.Time) ([]acp.ReaperSession, error) { return nil, nil } func (r *fakeAcpRepo) ListSessionUsage(_ context.Context, _ uuid.UUID, _ int32) ([]*acp.SessionUsageRecord, error) { return nil, nil } func (r *fakeAcpRepo) SummarizeAcpUsageByUser(_ context.Context, _, _ time.Time) ([]acp.AcpUsageUserRow, error) { return nil, nil } func (r *fakeAcpRepo) SummarizeAcpUsageByModel(_ context.Context, _, _ time.Time) ([]acp.AcpUsageModelRow, error) { return nil, nil } func (r *fakeAcpRepo) SummarizeAcpUsageByDay(_ context.Context, _, _ time.Time) ([]acp.AcpUsageDayRow, error) { return nil, nil } func (r *fakeAcpRepo) SessionRollup(_ context.Context, f acp.DashboardFilter) (acp.SessionRollup, error) { r.mu.Lock() defer r.mu.Unlock() r.rollupFilter = f return r.rollupResult, nil } func (r *fakeAcpRepo) SessionsByDay(_ context.Context, f acp.DashboardFilter) ([]acp.SessionDayRow, error) { r.mu.Lock() defer r.mu.Unlock() r.byDayFilter = f return r.byDayResult, nil } func (r *fakeAcpRepo) SessionTimeline(_ context.Context, sid uuid.UUID) ([]acp.SessionTimelineBucket, error) { r.mu.Lock() defer r.mu.Unlock() r.timelineSID = sid return r.timelineResult, nil } func (r *fakeAcpRepo) InsertTurn(_ context.Context, t *acp.Turn) (*acp.Turn, error) { return t, nil } func (r *fakeAcpRepo) MarkTurnCompleted(_ context.Context, _ uuid.UUID, _ int, _ string, _ int) error { return nil } func (r *fakeAcpRepo) MarkTurnAborted(_ context.Context, _ uuid.UUID, _ int) error { return nil } func (r *fakeAcpRepo) IncrementTurnUpdateCount(_ context.Context, _ uuid.UUID, _ int) error { return nil } func (r *fakeAcpRepo) GetLatestTurnBySession(_ context.Context, _ uuid.UUID) (*acp.Turn, error) { return nil, nil } func (r *fakeAcpRepo) ListTurnsBySession(_ context.Context, _ uuid.UUID) ([]*acp.Turn, error) { return nil, nil } func (r *fakeAcpRepo) UpdateSessionLastStopReason(_ context.Context, _ uuid.UUID, _ string) error { return nil } func (r *fakeAcpRepo) PurgeTurnsBefore(_ context.Context, _ time.Time) (int64, error) { return 0, nil } // fakeMCPTokenSvc satisfies mcp.TokenService for unit tests. type fakeMCPTokenSvc struct { mu sync.Mutex issued []mcp.IssueSystemTokenInput issueErr error revokedBySess []uuid.UUID revokeBySessFn func(ctx context.Context, sessionID uuid.UUID) error } func (f *fakeMCPTokenSvc) IssueAdmin(_ context.Context, _ mcp.IssueAdminInput) (mcp.IssueResult, error) { return mcp.IssueResult{}, nil } func (f *fakeMCPTokenSvc) IssueSystemToken(_ context.Context, _ mcp.IssueSystemTokenInput) (mcp.IssueResult, error) { return mcp.IssueResult{}, nil } func (f *fakeMCPTokenSvc) IssueSystemTokenWithTx(_ context.Context, _ pgx.Tx, in mcp.IssueSystemTokenInput) (mcp.IssueResult, error) { f.mu.Lock() defer f.mu.Unlock() if f.issueErr != nil { return mcp.IssueResult{}, f.issueErr } f.issued = append(f.issued, in) return mcp.IssueResult{ID: uuid.New(), Plaintext: "tok_" + in.ACPSessionID.String()[:8]}, nil } func (f *fakeMCPTokenSvc) Authenticate(_ context.Context, _ string) (*mcp.Token, error) { return nil, nil } func (f *fakeMCPTokenSvc) Revoke(_ context.Context, _, _ uuid.UUID) error { return nil } func (f *fakeMCPTokenSvc) RevokeBySession(_ context.Context, sessionID uuid.UUID) error { f.mu.Lock() defer f.mu.Unlock() f.revokedBySess = append(f.revokedBySess, sessionID) if f.revokeBySessFn != nil { return f.revokeBySessFn(context.Background(), sessionID) } return nil } func (f *fakeMCPTokenSvc) List(_ context.Context, _ *uuid.UUID, _ bool) ([]*mcp.Token, error) { return nil, nil } func (f *fakeMCPTokenSvc) GetByID(_ context.Context, _ uuid.UUID) (*mcp.Token, error) { return nil, nil } // fakeWsSvc satisfies workspace.WorkspaceService for Create-path tests. type fakeWsSvc struct { ws *workspace.Workspace } func (s *fakeWsSvc) Get(_ context.Context, _ workspace.Caller, _ uuid.UUID) (*workspace.Workspace, error) { return s.ws, nil } func (s *fakeWsSvc) Create(context.Context, workspace.Caller, string, workspace.CreateWorkspaceInput) (*workspace.Workspace, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) List(context.Context, workspace.Caller, string) ([]*workspace.Workspace, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) Update(context.Context, workspace.Caller, uuid.UUID, workspace.UpdateWorkspaceInput) (*workspace.Workspace, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) Delete(context.Context, workspace.Caller, uuid.UUID) error { return fmt.Errorf("unused") } func (s *fakeWsSvc) Sync(context.Context, workspace.Caller, uuid.UUID) (*workspace.Workspace, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) UpsertCredential(context.Context, workspace.Caller, uuid.UUID, workspace.UpsertCredentialInput) (*workspace.Credential, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) GetCredentialMeta(context.Context, workspace.Caller, uuid.UUID) (*workspace.Credential, error) { return nil, fmt.Errorf("unused") } func (s *fakeWsSvc) ListBranches(context.Context, workspace.Caller, uuid.UUID) ([]string, error) { return nil, fmt.Errorf("unused") } // ===== Tests ===== func TestSessionService_Create_IssuesSystemToken(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() pid := uuid.New() akID := uuid.New() repo := &fakeAcpRepo{ kinds: []*acp.AgentKind{ {ID: akID, Name: "claude", Enabled: true}, }, } wsSvc := &fakeWsSvc{ws: &workspace.Workspace{ ID: wsID, ProjectID: pid, DefaultBranch: "main", MainPath: "/tmp/main", }} pa := fakeProjectAccess{pj: &project.Project{ID: pid, Slug: "test-proj"}} mcpTokens := &fakeMCPTokenSvc{} sup := acp.NewSupervisor( repo, nil, nil, nil, nil, mcpTokens, nil, acp.SupervisorConfig{SpawnTimeout: 5 * time.Second, KillGrace: 1 * time.Second}, nil, // logger ) svc := acp.NewSessionService( repo, wsSvc, nil, nil, pa, nil, nil, sup, nil, acp.SessionServiceConfig{ MaxActiveGlobal: 10, MaxActivePerUser: 5, SystemTokenTTL: 24 * time.Hour, MCPPublicURL: "http://localhost:8080/mcp", }, mcpTokens, ) branch := "main" sess, err := svc.Create(context.Background(), acp.Caller{UserID: uid}, acp.CreateSessionInput{ WorkspaceID: wsID, AgentKindID: akID, Branch: &branch, }) require.NoError(t, err) require.NotNil(t, sess) assert.Equal(t, wsID, sess.WorkspaceID) assert.Equal(t, uid, sess.UserID) // Verify system token was issued inside the tx mcpTokens.mu.Lock() require.Len(t, mcpTokens.issued, 1) assert.Equal(t, uid, mcpTokens.issued[0].UserID) assert.Equal(t, sess.ID, mcpTokens.issued[0].ACPSessionID) assert.Equal(t, []uuid.UUID{pid}, mcpTokens.issued[0].ProjectIDs) assert.Equal(t, 24*time.Hour, mcpTokens.issued[0].ExpiresIn) mcpTokens.mu.Unlock() } func TestSessionService_Create_TokenIssueFails_RollsBack(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() pid := uuid.New() akID := uuid.New() repo := &fakeAcpRepo{ kinds: []*acp.AgentKind{ {ID: akID, Name: "claude", Enabled: true}, }, } wsSvc := &fakeWsSvc{ws: &workspace.Workspace{ ID: wsID, ProjectID: pid, DefaultBranch: "main", MainPath: "/tmp/main", }} pa := fakeProjectAccess{pj: &project.Project{ID: pid, Slug: "test-proj"}} mcpTokens := &fakeMCPTokenSvc{ issueErr: fmt.Errorf("token issue boom"), } sup := acp.NewSupervisor( repo, nil, nil, nil, nil, mcpTokens, nil, acp.SupervisorConfig{SpawnTimeout: 5 * time.Second, KillGrace: 1 * time.Second}, nil, ) svc := acp.NewSessionService( repo, wsSvc, nil, nil, pa, nil, nil, sup, nil, acp.SessionServiceConfig{ MaxActiveGlobal: 10, MaxActivePerUser: 5, SystemTokenTTL: 24 * time.Hour, MCPPublicURL: "http://localhost:8080/mcp", }, mcpTokens, ) branch := "main" sess, err := svc.Create(context.Background(), acp.Caller{UserID: uid}, acp.CreateSessionInput{ WorkspaceID: wsID, AgentKindID: akID, Branch: &branch, }) require.Error(t, err) assert.Nil(t, sess) assert.Contains(t, err.Error(), "token issue boom") // Verify no session was committed (InTx rolled back) repo.mu.Lock() assert.Empty(t, repo.sessions) repo.mu.Unlock() } func TestSessionService_Create_SpawnFails_MarksSessionCrashed(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() pid := uuid.New() akID := uuid.New() repo := &fakeAcpRepo{ kinds: []*acp.AgentKind{ // BinaryPath 不存在 → supervisor.Spawn 在 cmd.Start 早期失败, // 进程未注册、onExit 不会触发。 {ID: akID, Name: "claude", Enabled: true, BinaryPath: "/nonexistent/agent-binary"}, }, } wsSvc := &fakeWsSvc{ws: &workspace.Workspace{ ID: wsID, ProjectID: pid, DefaultBranch: "main", MainPath: "/tmp/main", }} pa := fakeProjectAccess{pj: &project.Project{ID: pid, Slug: "test-proj"}} mcpTokens := &fakeMCPTokenSvc{} sup := acp.NewSupervisor( repo, nil, nil, nil, nil, mcpTokens, nil, acp.SupervisorConfig{SpawnTimeout: 5 * time.Second, KillGrace: 1 * time.Second}, nil, ) svc := acp.NewSessionService( repo, wsSvc, nil, nil, pa, nil, nil, sup, nil, acp.SessionServiceConfig{ MaxActiveGlobal: 10, MaxActivePerUser: 5, SystemTokenTTL: 24 * time.Hour, MCPPublicURL: "http://localhost:8080/mcp", }, mcpTokens, ) branch := "main" sess, err := svc.Create(context.Background(), acp.Caller{UserID: uid}, acp.CreateSessionInput{ WorkspaceID: wsID, AgentKindID: akID, Branch: &branch, }) require.NoError(t, err) require.NotNil(t, sess) // async spawn 失败后:session 被守卫式标记为 crashed 且 last_error 非空 require.Eventually(t, func() bool { repo.mu.Lock() defer repo.mu.Unlock() for _, s := range repo.sessions { if s.ID == sess.ID { return s.Status == acp.SessionCrashed && s.LastError != nil && *s.LastError != "" } } return false }, 5*time.Second, 50*time.Millisecond, "session should be marked crashed with non-empty last_error") } func TestSessionService_Create_MutuallyExclusiveInputs(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() akID := uuid.New() svc := acp.NewSessionService( &fakeAcpRepo{}, nil, nil, nil, nil, nil, nil, nil, nil, acp.SessionServiceConfig{}, nil, ) branch := "feat/x" issueNum := 42 reqNum := 7 cases := []struct { name string in acp.CreateSessionInput }{ {"branch+issue", acp.CreateSessionInput{Branch: &branch, IssueNumber: &issueNum}}, {"branch+requirement", acp.CreateSessionInput{Branch: &branch, RequirementNumber: &reqNum}}, {"issue+requirement", acp.CreateSessionInput{IssueNumber: &issueNum, RequirementNumber: &reqNum}}, {"all three", acp.CreateSessionInput{Branch: &branch, IssueNumber: &issueNum, RequirementNumber: &reqNum}}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { tc.in.WorkspaceID = wsID tc.in.AgentKindID = akID _, err := svc.Create(context.Background(), acp.Caller{UserID: uid}, tc.in) require.Error(t, err) assert.Contains(t, err.Error(), "mutually exclusive") }) } } // ===== PermissionRequest(权限审批框架,测试桩) ===== func (f *fakeAcpRepo) InsertPermissionRequest(context.Context, *acp.PermissionRequest) (*acp.PermissionRequest, error) { return &acp.PermissionRequest{}, nil } func (f *fakeAcpRepo) GetPermissionRequestByID(context.Context, uuid.UUID) (*acp.PermissionRequest, error) { return nil, nil } func (f *fakeAcpRepo) ListPendingPermissionRequestsBySession(context.Context, uuid.UUID) ([]*acp.PermissionRequest, error) { return nil, nil } func (f *fakeAcpRepo) ListPendingPermissionRequestsByUser(context.Context, uuid.UUID) ([]*acp.PermissionRequest, error) { return nil, nil } func (f *fakeAcpRepo) DecidePermissionRequest(context.Context, uuid.UUID, acp.PermissionStatus, *string, *uuid.UUID) (*acp.PermissionRequest, bool, error) { return nil, false, nil } func (f *fakeAcpRepo) ExpirePendingPermissionRequestsBySession(context.Context, uuid.UUID) (int64, error) { return 0, nil } func (f *fakeAcpRepo) ExpireStalePermissionRequests(context.Context, time.Time) (int64, error) { return 0, nil } func TestSessionService_List_RequirementFilter(t *testing.T) { t.Parallel() uid := uuid.New() reqID := uuid.New() repo := &fakeAcpRepo{} repo.sessions = []*acp.Session{ {ID: uuid.New(), UserID: uid, RequirementID: &reqID}, {ID: uuid.New(), UserID: uid}, {ID: uuid.New(), UserID: uuid.New(), RequirementID: &reqID}, } svc := acp.NewSessionService( repo, nil, nil, nil, nil, nil, nil, nil, nil, acp.SessionServiceConfig{}, nil, ) // 普通用户:仅本人 + requirement 过滤 list, err := svc.List(context.Background(), acp.Caller{UserID: uid}, acp.SessionListFilter{RequirementID: &reqID}) require.NoError(t, err) require.Len(t, list, 1) // 非 admin 请求 all → 403 _, err = svc.List(context.Background(), acp.Caller{UserID: uid}, acp.SessionListFilter{All: true}) require.Error(t, err) // admin all + requirement 过滤:跨用户 2 条 list, err = svc.List(context.Background(), acp.Caller{UserID: uid, IsAdmin: true}, acp.SessionListFilter{All: true, RequirementID: &reqID}) require.NoError(t, err) require.Len(t, list, 2) } // TestSessionService_ResumeSession_ReacquiresMainWorktree 验证:在崩溃的 main-worktree // session 上 ResumeSession 会重新占用 main worktree(CAS)、重新签发 system token、把 // session 复位为 starting,然后尝试 spawn。binary 为空致 spawn 失败,但前置的占用/签发/ // 复位都应已发生。 func TestSessionService_ResumeSession_ReacquiresMainWorktree(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() pid := uuid.New() akID := uuid.New() sessID := uuid.New() exitCode := int32(1) crashed := &acp.Session{ ID: sessID, WorkspaceID: wsID, ProjectID: pid, AgentKindID: akID, UserID: uid, Branch: "main", CwdPath: "/tmp/main", IsMainWorktree: true, Status: acp.SessionCrashed, ExitCode: &exitCode, } repo := &fakeAcpRepo{ kinds: []*acp.AgentKind{{ID: akID, Name: "claude", Enabled: true}}, sessions: []*acp.Session{crashed}, } mcpTokens := &fakeMCPTokenSvc{} sup := acp.NewSupervisor( repo, nil, nil, nil, nil, mcpTokens, nil, acp.SupervisorConfig{SpawnTimeout: 2 * time.Second, KillGrace: time.Second}, nil, ) svc := acp.NewSessionService( repo, &fakeWsSvc{ws: &workspace.Workspace{ID: wsID, ProjectID: pid, DefaultBranch: "main", MainPath: "/tmp/main"}}, nil, nil, fakeProjectAccess{pj: &project.Project{ID: pid, Slug: "p"}}, nil, nil, sup, nil, acp.SessionServiceConfig{MaxActiveGlobal: 10, MaxActivePerUser: 5, SystemTokenTTL: time.Hour, MCPPublicURL: "http://x/mcp"}, mcpTokens, ) tr, ok := svc.(acp.TurnRunner) require.True(t, ok, "sessionService must implement acp.TurnRunner") // spawn 会失败(binary 空),ResumeSession 返回错误是预期的。 _, _ = tr.ResumeSession(context.Background(), acp.Caller{UserID: uid}, sessID) // 但 token 应已重新签发(占用 + InTx 复位在 spawn 之前)。 mcpTokens.mu.Lock() issuedCount := len(mcpTokens.issued) mcpTokens.mu.Unlock() require.GreaterOrEqual(t, issuedCount, 1, "resume should re-issue a system token") assert.Equal(t, sessID, mcpTokens.issued[0].ACPSessionID) } // TestSessionService_ResumeSession_ActiveSessionNoOp 验证:对已活跃 session 调用 // ResumeSession 直接返回该 session,不重新占用/签发。 func TestSessionService_ResumeSession_ActiveSessionNoOp(t *testing.T) { t.Parallel() uid := uuid.New() wsID := uuid.New() pid := uuid.New() akID := uuid.New() sessID := uuid.New() active := &acp.Session{ ID: sessID, WorkspaceID: wsID, ProjectID: pid, AgentKindID: akID, UserID: uid, Branch: "main", CwdPath: "/tmp/main", IsMainWorktree: true, Status: acp.SessionRunning, } repo := &fakeAcpRepo{ kinds: []*acp.AgentKind{{ID: akID, Name: "claude", Enabled: true}}, sessions: []*acp.Session{active}, } mcpTokens := &fakeMCPTokenSvc{} sup := acp.NewSupervisor(repo, nil, nil, nil, nil, mcpTokens, nil, acp.SupervisorConfig{SpawnTimeout: time.Second, KillGrace: time.Second}, nil) svc := acp.NewSessionService(repo, &fakeWsSvc{ws: &workspace.Workspace{ID: wsID, ProjectID: pid, DefaultBranch: "main", MainPath: "/tmp/main"}}, nil, nil, fakeProjectAccess{pj: &project.Project{ID: pid, Slug: "p"}}, nil, nil, sup, nil, acp.SessionServiceConfig{MaxActiveGlobal: 10, MaxActivePerUser: 5, SystemTokenTTL: time.Hour, MCPPublicURL: "http://x/mcp"}, mcpTokens) tr := svc.(acp.TurnRunner) out, err := tr.ResumeSession(context.Background(), acp.Caller{UserID: uid}, sessID) require.NoError(t, err) require.NotNil(t, out) assert.Equal(t, sessID, out.ID) mcpTokens.mu.Lock() defer mcpTokens.mu.Unlock() assert.Len(t, mcpTokens.issued, 0, "active session resume must not re-issue tokens") } // ===== run-history dashboard ===== func newDashboardSvc(repo *fakeAcpRepo) acp.SessionService { return acp.NewSessionService(repo, nil, nil, nil, nil, nil, nil, nil, nil, acp.SessionServiceConfig{}, nil) } func TestDashboard_NonAdmin_ScopedToCaller(t *testing.T) { t.Parallel() uid := uuid.New() repo := &fakeAcpRepo{ rollupResult: acp.SessionRollup{Total: 4, Succeeded: 3, Crashed: 1, TotalCostUSD: 1.5, TotalTokens: 900}, byDayResult: []acp.SessionDayRow{{Day: time.Now().UTC(), Total: 4, Succeeded: 3, Crashed: 1, TotalCostUSD: 1.5}}, } svc := newDashboardSvc(repo) res, err := svc.Dashboard(context.Background(), acp.Caller{UserID: uid}, acp.DashboardInput{}) require.NoError(t, err) require.NotNil(t, res) // 非 admin 强制按 caller scope。 require.NotNil(t, repo.rollupFilter.UserID) assert.Equal(t, uid, *repo.rollupFilter.UserID) require.NotNil(t, repo.byDayFilter.UserID) assert.Equal(t, uid, *repo.byDayFilter.UserID) // 默认时间窗为半开且 from 早于 to。 assert.True(t, repo.rollupFilter.From.Before(repo.rollupFilter.To)) // rollup 透传 + 成功率计算。 assert.Equal(t, int64(4), res.Rollup.Total) assert.InDelta(t, 0.75, res.Rollup.SuccessRate(), 1e-9) assert.InDelta(t, 1.5, res.Rollup.TotalCostUSD, 1e-9) assert.Len(t, res.ByDay, 1) } func TestDashboard_Admin_NoUserScope_WithFilters(t *testing.T) { t.Parallel() pid := uuid.New() reqID := uuid.New() from := time.Now().Add(-48 * time.Hour).UTC() to := time.Now().UTC() repo := &fakeAcpRepo{rollupResult: acp.SessionRollup{Total: 10, Succeeded: 10}} svc := newDashboardSvc(repo) res, err := svc.Dashboard(context.Background(), acp.Caller{UserID: uuid.New(), IsAdmin: true}, acp.DashboardInput{ProjectID: &pid, RequirementID: &reqID, From: from, To: to}) require.NoError(t, err) // admin → 无 user scope;project/requirement/时间窗透传。 assert.Nil(t, repo.rollupFilter.UserID) require.NotNil(t, repo.rollupFilter.ProjectID) assert.Equal(t, pid, *repo.rollupFilter.ProjectID) require.NotNil(t, repo.rollupFilter.RequirementID) assert.Equal(t, reqID, *repo.rollupFilter.RequirementID) assert.Equal(t, from, repo.rollupFilter.From) assert.Equal(t, to, repo.rollupFilter.To) assert.InDelta(t, 1.0, res.Rollup.SuccessRate(), 1e-9) } func TestDashboard_InvalidWindow(t *testing.T) { t.Parallel() to := time.Now().UTC() from := to.Add(time.Hour) // from after to svc := newDashboardSvc(&fakeAcpRepo{}) _, err := svc.Dashboard(context.Background(), acp.Caller{UserID: uuid.New()}, acp.DashboardInput{From: from, To: to}) require.Error(t, err) } func TestTimeline_OwnerAccess(t *testing.T) { t.Parallel() uid := uuid.New() sid := uuid.New() buckets := []acp.SessionTimelineBucket{ {Direction: "out", RPCKind: "request", Method: "session/prompt", EventCount: 2}, {Direction: "in", RPCKind: "notification", Method: "session/update", EventCount: 5}, } repo := &fakeAcpRepo{timelineResult: buckets} repo.sessions = []*acp.Session{{ID: sid, UserID: uid}} svc := newDashboardSvc(repo) got, err := svc.Timeline(context.Background(), acp.Caller{UserID: uid}, sid) require.NoError(t, err) assert.Equal(t, sid, repo.timelineSID) require.Len(t, got, 2) assert.Equal(t, "session/prompt", got[0].Method) } func TestTimeline_NonOwner_NotFound(t *testing.T) { t.Parallel() sid := uuid.New() repo := &fakeAcpRepo{} repo.sessions = []*acp.Session{{ID: sid, UserID: uuid.New()}} svc := newDashboardSvc(repo) _, err := svc.Timeline(context.Background(), acp.Caller{UserID: uuid.New()}, sid) require.Error(t, err) // Get 拒绝 → 不泄漏会话 }