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

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 拒绝 → 不泄漏会话
}