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

// session_service.go 实现 SessionService。Create 走完整事务（worktree
// acquire + supervisor.Spawn）；Get/List/Terminate 是配套的简单读写。
package acp

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"

	"github.com/yan1h/agent-coding-workflow/internal/audit"
	"github.com/yan1h/agent-coding-workflow/internal/brief"
	"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
	"github.com/yan1h/agent-coding-workflow/internal/mcp"
	"github.com/yan1h/agent-coding-workflow/internal/project"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

// SessionServiceConfig 是 service 的限流 + MCP 配置（从 AcpConfig / MCPConfig 派生）。
type SessionServiceConfig struct {
	MaxActiveGlobal  int
	MaxActivePerUser int
	SystemTokenTTL   time.Duration
	MCPPublicURL     string
	BriefTokenBudget int // 初始 prompt 简报的 token 硬上限（<=0 用 brief 默认值）
}

// ProjectAccess 是 acp 模块对 project 模块的窄接口；adapter 在 app.go 实现。
type ProjectAccess interface {
	GetProjectByWorkspace(ctx context.Context, wsID uuid.UUID) (*project.Project, error)
	GetIssueByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Issue, error)
	GetRequirementByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Requirement, error)
}

// ArtifactAccess 是 acp 模块拉取某 requirement 各阶段最新产物的窄接口；adapter 在
// app.go 实现（基于 project.Repository.ListArtifactsByRequirement，按阶段取最大版本）。
type ArtifactAccess interface {
	ListLatestArtifacts(ctx context.Context, requirementID uuid.UUID) ([]*project.Artifact, error)
}

type sessionService struct {
	repo      Repository
	wsSvc     workspace.WorkspaceService
	wtSvc     workspace.WorktreeService
	wsRepo    workspace.Repository
	pa        ProjectAccess
	aa        ArtifactAccess // 可为 nil（无产物注入）
	composer  brief.Composer // 可为 nil（退回 raw passthrough）
	sup       *Supervisor
	audit     audit.Recorder
	cfg       SessionServiceConfig
	mcpTokens mcp.TokenService // spec §6.1 — system token issuance
}

// NewSessionService 构造 SessionService。aa / composer 可为 nil：缺任一时
// initial prompt 退回原始透传（不组装简报），保持旧行为。
func NewSessionService(repo Repository, wsSvc workspace.WorkspaceService,
	wtSvc workspace.WorktreeService, wsRepo workspace.Repository,
	pa ProjectAccess, aa ArtifactAccess, composer brief.Composer,
	sup *Supervisor, rec audit.Recorder,
	cfg SessionServiceConfig, mcpTokens mcp.TokenService) SessionService {
	return &sessionService{
		repo: repo, wsSvc: wsSvc, wtSvc: wtSvc, wsRepo: wsRepo,
		pa: pa, aa: aa, composer: composer, sup: sup, audit: rec,
		cfg: cfg, mcpTokens: mcpTokens,
	}
}

// ResolveBranch 是 spec §8.1 的纯函数实现：根据 CreateSessionInput 决定
// (branch, isMainWorktree)。允许独立单测，不依赖 service 状态。
//
// 规则（按优先级）：
//  1. in.Branch 显式：直接用；isMain = (== ws.DefaultBranch)
//  2. in.IssueNumber：拉 issue，校验 workspace 关联，命名 "issue/<slug>-<num>"
//  3. in.RequirementNumber：拉 requirement，校验 workspace，命名 "req/<slug>-<num>"
//  4. 都没给：fallback "acp-auto/<8-hex>"，isMain = false
func ResolveBranch(ctx context.Context, ws *workspace.Workspace, sessionID uuid.UUID,
	in CreateSessionInput, projectSlug string, pa ProjectAccess) (string, bool, error) {

	if in.Branch != nil && *in.Branch != "" {
		br := *in.Branch
		return br, br == ws.DefaultBranch, nil
	}

	if in.IssueNumber != nil {
		iss, err := pa.GetIssueByNumber(ctx, ws.ProjectID, *in.IssueNumber)
		if err != nil {
			return "", false, err
		}
		if iss.WorkspaceID != nil && *iss.WorkspaceID != ws.ID {
			return "", false, errs.New(errs.CodeInvalidInput,
				fmt.Sprintf("issue #%d already linked to other workspace", *in.IssueNumber))
		}
		return fmt.Sprintf("issue/%s-%d", projectSlug, iss.Number), false, nil
	}

	if in.RequirementNumber != nil {
		req, err := pa.GetRequirementByNumber(ctx, ws.ProjectID, *in.RequirementNumber)
		if err != nil {
			return "", false, err
		}
		if req.WorkspaceID != nil && *req.WorkspaceID != ws.ID {
			return "", false, errs.New(errs.CodeInvalidInput,
				fmt.Sprintf("requirement #%d already linked to other workspace", *in.RequirementNumber))
		}
		return fmt.Sprintf("req/%s-%d", projectSlug, req.Number), false, nil
	}

	short := strings.ReplaceAll(sessionID.String(), "-", "")
	if len(short) > 8 {
		short = short[:8]
	}
	return fmt.Sprintf("acp-auto/%s", short), false, nil
}

// Create 执行完整 session 创建事务：
//  1. workspace 鉴权（wsSvc.Get）
//  2. agent kind 校验（enabled?）
//  3. 全局 + 单用户限流
//  4. 分支解析（ResolveBranch + project 查询）
//  5. worktree 占用（main → repo.AcquireMainWorktree；
//     子分支 → wtSvc.List 找已存在或 Create + Acquire，holder = "session:<sid>"）
//  6. issueID/requirementID 反查（best-effort）
//  7. INSERT acp_sessions
//  8. audit.Record
//  9. async supervisor.Spawn（失败回滚 worktree）
//  10. async initial_prompt 转发（等握手完成）
func (s *sessionService) Create(ctx context.Context, c Caller, in CreateSessionInput) (*Session, error) {
	// 0. 互斥校验：branch / issue_number / requirement_number 三选一或全空（spec §8.1）。
	// service 层统一拦截，HTTP / MCP 入口共用。
	specified := 0
	if in.Branch != nil && *in.Branch != "" {
		specified++
	}
	if in.IssueNumber != nil {
		specified++
	}
	if in.RequirementNumber != nil {
		specified++
	}
	if specified > 1 {
		return nil, errs.New(errs.CodeInvalidInput,
			"branch / issue_number / requirement_number are mutually exclusive; specify at most one")
	}

	// 1. workspace + authz
	wsCaller := workspace.Caller{UserID: c.UserID, IsAdmin: c.IsAdmin}
	ws, err := s.wsSvc.Get(ctx, wsCaller, in.WorkspaceID)
	if err != nil {
		return nil, err
	}

	// 2. agent kind
	kind, err := s.repo.GetAgentKindByID(ctx, in.AgentKindID)
	if err != nil {
		return nil, err
	}
	if !kind.Enabled {
		return nil, errs.New(errs.CodeAcpAgentKindDisabled, "agent kind disabled")
	}

	// 3. 限流
	nGlobal, _ := s.repo.CountActiveSessions(ctx)
	if int(nGlobal) >= s.cfg.MaxActiveGlobal {
		return nil, errs.New(errs.CodeAcpSessionQuotaExceeded, "global session quota exceeded")
	}
	nUser, _ := s.repo.CountActiveSessionsByUser(ctx, c.UserID)
	if int(nUser) >= s.cfg.MaxActivePerUser {
		return nil, errs.New(errs.CodeAcpSessionQuotaExceeded, "per-user session quota exceeded")
	}

	// 4. 分支解析
	pj, err := s.pa.GetProjectByWorkspace(ctx, ws.ID)
	if err != nil {
		return nil, err
	}
	sessionID := uuid.New()
	branch, isMain, err := ResolveBranch(ctx, ws, sessionID, in, pj.Slug, s.pa)
	if err != nil {
		return nil, err
	}

	// 5. worktree 占用
	var (
		cwdPath    string
		worktreeID *uuid.UUID
	)
	sessCaller := workspace.Caller{UserID: c.UserID, IsAdmin: c.IsAdmin, SessionID: &sessionID}
	if isMain {
		ok, err := s.repo.AcquireMainWorktree(ctx, sessionID, ws.ID)
		if err != nil {
			return nil, err
		}
		if !ok {
			return nil, errs.New(errs.CodeAcpSessionBranchConflict, "main worktree already in use")
		}
		cwdPath = ws.MainPath
	} else {
		wt, err := s.findOrCreateWorktree(ctx, sessCaller, ws.ID, branch)
		if err != nil {
			return nil, err
		}
		if _, err := s.wtSvc.Acquire(ctx, sessCaller, wt.ID); err != nil {
			return nil, err
		}
		cwdPath = wt.Path
		wid := wt.ID
		worktreeID = &wid
	}

	// 6. PM 关联（best-effort：拿不到 ID 不阻塞 session 创建）。
	// 同时保留解析出的 issue/requirement 对象，供初始 prompt 组装简报复用。
	var (
		issueID, reqID *uuid.UUID
		pmIssue        *project.Issue
		pmReq          *project.Requirement
	)
	if in.IssueNumber != nil {
		if iss, _ := s.pa.GetIssueByNumber(ctx, ws.ProjectID, *in.IssueNumber); iss != nil {
			id := iss.ID
			issueID = &id
			pmIssue = iss
		}
	}
	if in.RequirementNumber != nil {
		if req, _ := s.pa.GetRequirementByNumber(ctx, ws.ProjectID, *in.RequirementNumber); req != nil {
			id := req.ID
			reqID = &id
			pmReq = req
		}
	}

	// 7. INSERT acp_sessions + system MCP token（spec §6.1 atomic guarantee）
	// 预算快照：当前无 session 级覆盖入口，故直接继承 agent-kind 默认上限。快照到
	// session 行使 supervisor / reaper 读取自洽，不必每次回查 agent-kind。
	sess := &Session{
		ID: sessionID, WorkspaceID: ws.ID, ProjectID: ws.ProjectID,
		AgentKindID: kind.ID, UserID: c.UserID,
		IssueID: issueID, RequirementID: reqID,
		Branch: branch, CwdPath: cwdPath, IsMainWorktree: isMain,
		Status:                    SessionStarting,
		OrchestratorStepID:        in.OrchestratorStepID,
		BudgetMaxCostUSD:          kind.MaxCostUSD,
		BudgetMaxTokens:           kind.MaxTokens,
		BudgetMaxWallClockSeconds: kind.MaxWallClockSeconds,
	}
	var (
		out      *Session
		tokenRes mcp.IssueResult
	)
	txErr := s.repo.InTx(ctx, func(txCtx context.Context, tx pgx.Tx) error {
		inserted, err := s.repo.WithTx(tx).InsertSession(txCtx, sess)
		if err != nil {
			return err
		}
		out = inserted

		res, err := s.mcpTokens.IssueSystemTokenWithTx(txCtx, tx, mcp.IssueSystemTokenInput{
			UserID:       c.UserID,
			ACPSessionID: inserted.ID,
			ProjectIDs:   []uuid.UUID{ws.ProjectID},
			ExpiresIn:    s.cfg.SystemTokenTTL,
		})
		if err != nil {
			return err
		}
		tokenRes = res
		return nil
	})
	if txErr != nil {
		s.releaseOnFailure(ctx, sess, sessCaller, worktreeID)
		return nil, txErr
	}

	// 8. audit
	s.recordAudit(ctx, c, "acp.session.create", out.ID.String(), map[string]any{
		"agent_kind":     kind.Name,
		"branch":         branch,
		"cwd_path":       cwdPath,
		"issue_id":       issueID,
		"requirement_id": reqID,
	})

	// 9. async spawn（失败时回滚 worktree + revoke MCP token）
	extraEnv := map[string]string{
		EnvMCPToken: tokenRes.Plaintext,
		EnvMCPURL:   s.cfg.MCPPublicURL,
	}
	go func() {
		spawnCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
		defer cancel()
		if _, err := s.sup.Spawn(spawnCtx, out, kind, extraEnv); err != nil {
			// 守卫式标记：Spawn 早期失败（进程未注册、onExit 不会触发）时把
			// starting 收尾为 crashed；handshake 失败路径 onExit 已写终态，此处 no-op。
			if _, merr := s.repo.MarkSessionFailedIfActive(spawnCtx, out.ID, err.Error()); merr != nil {
				s.sup.log.Error("acp.spawn_failed.mark_session",
					"session_id", out.ID, "err", merr.Error())
			}
			_ = s.mcpTokens.RevokeBySession(spawnCtx, out.ID)
			s.releaseOnFailure(spawnCtx, out, sessCaller, worktreeID)
		}
	}()

	// 10. initial prompt：握手成功后异步转发。先在请求路径同步组装服务端简报
	// （需求/Issue + 阶段产物 + 仓库定向 + 操作者自由文本），失败/无 PM 上下文时
	// 退回原始透传，保持旧行为。
	if in.InitialPrompt != nil && *in.InitialPrompt != "" {
		text := s.composeInitialPrompt(ctx, c, out, pj, pmReq, pmIssue, *in.InitialPrompt)
		go s.sendInitialPrompt(out.ID, text)
	}

	return out, nil
}

// composeInitialPrompt 组装 ACP session 的初始 prompt 简报。无 composer 或无 PM 上下文
// （既无 requirement 也无 issue）时返回原始 userPrompt（raw passthrough）。组装失败也
// 静默退回原始文本，绝不阻塞 session。成功组装后写一条 audit。
func (s *sessionService) composeInitialPrompt(ctx context.Context, c Caller, sess *Session,
	pj *project.Project, req *project.Requirement, iss *project.Issue, userPrompt string) string {

	if s.composer == nil || (req == nil && iss == nil) {
		return userPrompt
	}

	var artifacts []*project.Artifact
	if req != nil && s.aa != nil {
		if arts, err := s.aa.ListLatestArtifacts(ctx, req.ID); err == nil {
			artifacts = arts
		}
	}

	res, err := s.composer.ComposeTaskBrief(ctx, brief.BriefInput{
		Project:     pj,
		Requirement: req,
		Issue:       iss,
		Artifacts:   artifacts,
		RepoCwd:     sess.CwdPath,
		Branch:      sess.Branch,
		UserPrompt:  userPrompt,
		TokenBudget: s.cfg.BriefTokenBudget,
		Kind:        brief.KindACPSession,
	})
	if err != nil || res.Text == "" {
		return userPrompt
	}

	s.recordAudit(ctx, c, "acp.session.brief_composed", sess.ID.String(), map[string]any{
		"requirement_id":    idOrNil(req),
		"issue_id":          issueIDOrNil(iss),
		"artifact_versions": artifactVersions(artifacts),
		"brief_tokens":      res.Tokens,
		"truncated":         res.Truncated,
		"sections":          res.Sections,
	})
	return res.Text
}

func idOrNil(req *project.Requirement) *uuid.UUID {
	if req == nil {
		return nil
	}
	id := req.ID
	return &id
}

func issueIDOrNil(iss *project.Issue) *uuid.UUID {
	if iss == nil {
		return nil
	}
	id := iss.ID
	return &id
}

// artifactVersions 把产物列表归约为 {phase: version} 映射（每阶段取出现的最大版本），
// 仅用于 audit 元数据。
func artifactVersions(arts []*project.Artifact) map[string]int {
	out := make(map[string]int, len(arts))
	for _, a := range arts {
		if a == nil {
			continue
		}
		if v, ok := out[string(a.Phase)]; !ok || a.Version > v {
			out[string(a.Phase)] = a.Version
		}
	}
	return out
}

// findOrCreateWorktree 在指定 workspace 内查找 branch 同名 worktree，
// 不存在则创建。caller 已通过 wsSvc.Get 完成鉴权。
func (s *sessionService) findOrCreateWorktree(ctx context.Context, c workspace.Caller, wsID uuid.UUID, branch string) (*workspace.Worktree, error) {
	wts, err := s.wtSvc.List(ctx, c, wsID)
	if err != nil {
		return nil, err
	}
	for _, wt := range wts {
		if wt.Branch == branch {
			return wt, nil
		}
	}
	return s.wtSvc.Create(ctx, c, wsID, workspace.CreateWorktreeInput{Branch: branch})
}

// releaseOnFailure 在 INSERT / Spawn 失败时回滚 worktree 占用。幂等且静默。
func (s *sessionService) releaseOnFailure(ctx context.Context, sess *Session, c workspace.Caller, worktreeID *uuid.UUID) {
	if sess.IsMainWorktree {
		_, _ = s.repo.ReleaseMainWorktree(ctx, sess.WorkspaceID, sess.ID)
		return
	}
	if worktreeID != nil {
		_, _ = s.wtSvc.Release(ctx, c, *worktreeID)
	}
}

// sendInitialPrompt 等待 supervisor 完成 handshake（最多 30s），随后转发一条
// session/prompt 给 agent。轮询 200ms 一次拉 relay + agent_session_id；任一缺失
// 继续等待。超时后静默放弃。
func (s *sessionService) sendInitialPrompt(sid uuid.UUID, text string) {
	deadline := time.After(30 * time.Second)
	tick := time.NewTicker(200 * time.Millisecond)
	defer tick.Stop()
	for {
		select {
		case <-deadline:
			return
		case <-tick.C:
		}
		relay := s.sup.GetRelay(sid)
		if relay == nil {
			continue
		}
		sess, _ := s.repo.GetSessionByID(context.Background(), sid)
		if sess == nil || sess.AgentSessionID == nil {
			continue
		}
		idJSON, _ := json.Marshal("client-init")
		params, _ := json.Marshal(map[string]any{
			"sessionId": *sess.AgentSessionID,
			"prompt":    []map[string]any{{"type": "text", "text": text}},
		})
		msg := &Message{
			JSONRPC: "2.0", ID: idJSON, Method: "session/prompt", Params: params,
		}
		// 回合相关联：在发送前登记回合 id（"client-init"），使响应的 stopReason 可被
		// 相关联到这个回合。tracker 可能为 nil（部分测试）——StartTurn 已做 nil 守卫。
		if tr := relay.Tracker(); tr != nil {
			if _, err := tr.StartTurn(context.Background(), "client-init"); err != nil {
				slog.Warn("acp.session.start_turn", "session_id", sid, "err", err.Error())
			}
		}
		_ = relay.SendClient(msg)
		return
	}
}

// 编译期断言：sessionService 同时实现 SessionService 与 TurnRunner（编排器依赖后者）。
var _ TurnRunner = (*sessionService)(nil)

// promptResult 是 session/prompt 响应 Result 中我们关心的字段。
type promptResult struct {
	StopReason string `json:"stopReason"`
}

// SendPromptAndWait 发送一条 session/prompt 并阻塞等待回合完成，返回 agent 上报的原始
// stopReason。这是编排器驱动一个阻塞回合的核心原语：先轮询等待 supervisor handshake
// 完成（relay + agent_session_id 就绪，复用 sendInitialPrompt 的就绪检测），随后走
// server-initiated relay.Call('session/prompt') 取响应里的 stopReason。
//
// ctx 控制整体超时（编排器以 cfg.TurnTimeout 约束，须 < jobReaper LeaseTimeout）。
func (s *sessionService) SendPromptAndWait(ctx context.Context, sessionID uuid.UUID, prompt string) (string, error) {
	relay, agentSessionID, err := s.waitForRelayReady(ctx, sessionID)
	if err != nil {
		return "", err
	}

	params := map[string]any{
		"sessionId": agentSessionID,
		"prompt":    []map[string]any{{"type": "text", "text": prompt}},
	}
	resp, err := relay.Call(ctx, "session/prompt", params)
	if err != nil {
		return "", errs.Wrap(err, errs.CodeInternal, "session/prompt call")
	}
	var pr promptResult
	if len(resp.Result) > 0 {
		if uerr := json.Unmarshal(resp.Result, &pr); uerr != nil {
			return "", errs.Wrap(uerr, errs.CodeInternal, "parse session/prompt result")
		}
	}
	if pr.StopReason != "" {
		// 落库最近一次 stopReason（best-effort），供观测/网关复用。
		_ = s.repo.UpdateSessionLastStopReason(context.WithoutCancel(ctx), sessionID, pr.StopReason)
	}
	return pr.StopReason, nil
}

// waitForRelayReady 轮询直到 relay 与 agent_session_id 同时就绪或 ctx 截止。
// 200ms 一次，与 sendInitialPrompt 一致。
func (s *sessionService) waitForRelayReady(ctx context.Context, sessionID uuid.UUID) (*Relay, string, error) {
	tick := time.NewTicker(200 * time.Millisecond)
	defer tick.Stop()
	for {
		relay := s.sup.GetRelay(sessionID)
		if relay != nil {
			sess, _ := s.repo.GetSessionByID(ctx, sessionID)
			if sess != nil && sess.AgentSessionID != nil {
				return relay, *sess.AgentSessionID, nil
			}
		}
		select {
		case <-ctx.Done():
			return nil, "", ctx.Err()
		case <-tick.C:
		}
	}
}

// ResumeSession 在崩溃/退出 session 的同一 cwd/worktree 上重新拉起 agent。复用原
// session 的 workspace/branch/cwd/main 标记与编排器 step 关联：重新占用 worktree
// （main → AcquireMainWorktree CAS；子 worktree → 按 holder Acquire），签发新 system
// token，再 sup.Spawn。Spawn 失败时回滚 worktree（releaseOnFailure）。
//
// 返回的是被复用并重置为 starting 状态的同一行 session（同 ID），供编排器随后
// SendPromptAndWait。仅供编排器内部调用（caller 为 run.owner）。
func (s *sessionService) ResumeSession(ctx context.Context, c Caller, sessionID uuid.UUID) (*Session, error) {
	sess, err := s.repo.GetSessionByID(ctx, sessionID)
	if err != nil {
		return nil, err
	}
	if sess.Status.IsActive() {
		// 已经活跃：无需 resume，直接返回。
		return sess, nil
	}

	kind, err := s.repo.GetAgentKindByID(ctx, sess.AgentKindID)
	if err != nil {
		return nil, err
	}

	sessCaller := workspace.Caller{UserID: c.UserID, IsAdmin: c.IsAdmin, SessionID: &sess.ID}
	if sess.IsMainWorktree {
		ok, aerr := s.repo.AcquireMainWorktree(ctx, sess.ID, sess.WorkspaceID)
		if aerr != nil {
			return nil, aerr
		}
		if !ok {
			return nil, errs.New(errs.CodeAcpSessionBranchConflict, "main worktree in use; cannot resume")
		}
	} else {
		wt, ferr := s.findOrCreateWorktree(ctx, sessCaller, sess.WorkspaceID, sess.Branch)
		if ferr != nil {
			return nil, ferr
		}
		if _, aerr := s.wtSvc.Acquire(ctx, sessCaller, wt.ID); aerr != nil {
			return nil, aerr
		}
	}

	// 重新签发 system token + 标记 session 回到 starting（清空上次终态字段）。
	var (
		tokenRes mcp.IssueResult
	)
	txErr := s.repo.InTx(ctx, func(txCtx context.Context, tx pgx.Tx) error {
		if rerr := s.repo.WithTx(tx).ResetSessionForResume(txCtx, sess.ID); rerr != nil {
			return rerr
		}
		res, ierr := s.mcpTokens.IssueSystemTokenWithTx(txCtx, tx, mcp.IssueSystemTokenInput{
			UserID:       c.UserID,
			ACPSessionID: sess.ID,
			ProjectIDs:   []uuid.UUID{sess.ProjectID},
			ExpiresIn:    s.cfg.SystemTokenTTL,
		})
		if ierr != nil {
			return ierr
		}
		tokenRes = res
		return nil
	})
	if txErr != nil {
		s.releaseOnFailure(ctx, sess, sessCaller, nil)
		return nil, txErr
	}
	sess.Status = SessionStarting

	extraEnv := map[string]string{
		EnvMCPToken: tokenRes.Plaintext,
		EnvMCPURL:   s.cfg.MCPPublicURL,
	}
	if _, serr := s.sup.Spawn(ctx, sess, kind, extraEnv); serr != nil {
		if _, merr := s.repo.MarkSessionFailedIfActive(ctx, sess.ID, serr.Error()); merr != nil {
			s.sup.log.Error("acp.resume.mark_session", "session_id", sess.ID, "err", merr.Error())
		}
		_ = s.mcpTokens.RevokeBySession(ctx, sess.ID)
		s.releaseOnFailure(ctx, sess, sessCaller, nil)
		return nil, serr
	}

	s.recordAudit(ctx, c, "acp.session.resume", sess.ID.String(), map[string]any{
		"branch":   sess.Branch,
		"cwd_path": sess.CwdPath,
	})
	return sess, nil
}

func (s *sessionService) Get(ctx context.Context, c Caller, id uuid.UUID) (*Session, error) {
	sess, err := s.repo.GetSessionByID(ctx, id)
	if err != nil {
		return nil, err
	}
	if !c.IsAdmin && sess.UserID != c.UserID {
		return nil, errs.New(errs.CodeAcpSessionNotFound, "session not found")
	}
	return sess, nil
}

func (s *sessionService) List(ctx context.Context, c Caller, f SessionListFilter) ([]*Session, error) {
	if f.All && !c.IsAdmin {
		return nil, errs.New(errs.CodeForbidden, "admin only")
	}
	if f.All {
		return s.repo.ListAllSessions(ctx, f.RequirementID)
	}
	return s.repo.ListSessionsByUser(ctx, c.UserID, f.RequirementID)
}

func (s *sessionService) Terminate(ctx context.Context, c Caller, id uuid.UUID) error {
	sess, err := s.Get(ctx, c, id)
	if err != nil {
		return err
	}
	if !sess.Status.IsActive() {
		return nil
	}
	s.sup.Kill(ctx, id, 5*time.Second)
	return nil
}

// defaultDashboardWindow 是 run-history 汇总在调用方未指定时间窗时的默认回溯。
const defaultDashboardWindow = 30 * 24 * time.Hour

// Dashboard 返回 run-history 汇总 + 按天序列。非 admin 强制按 caller scope。
func (s *sessionService) Dashboard(ctx context.Context, c Caller, in DashboardInput) (*DashboardResult, error) {
	to := in.To
	if to.IsZero() {
		to = time.Now().UTC()
	}
	from := in.From
	if from.IsZero() {
		from = to.Add(-defaultDashboardWindow)
	}
	if !from.Before(to) {
		return nil, errs.New(errs.CodeInvalidInput, "from must be before to")
	}
	f := DashboardFilter{
		ProjectID:     in.ProjectID,
		RequirementID: in.RequirementID,
		From:          from,
		To:            to,
	}
	// 非 admin 只能看自己的会话：UserID 固定为 caller。admin 留空（全量）。
	if !c.IsAdmin {
		uid := c.UserID
		f.UserID = &uid
	}
	rollup, err := s.repo.SessionRollup(ctx, f)
	if err != nil {
		return nil, err
	}
	byDay, err := s.repo.SessionsByDay(ctx, f)
	if err != nil {
		return nil, err
	}
	return &DashboardResult{Rollup: rollup, ByDay: byDay, From: from, To: to}, nil
}

// Timeline 返回单会话事件时间线（先经 Get 做 owner/admin 访问校验）。
func (s *sessionService) Timeline(ctx context.Context, c Caller, sessionID uuid.UUID) ([]SessionTimelineBucket, error) {
	if _, err := s.Get(ctx, c, sessionID); err != nil {
		return nil, err
	}
	return s.repo.SessionTimeline(ctx, sessionID)
}

func (s *sessionService) recordAudit(ctx context.Context, c Caller, action, targetID string, meta map[string]any) {
	if s.audit == nil {
		return
	}
	uid := c.UserID
	_ = s.audit.Record(context.WithoutCancel(ctx), audit.Entry{
		UserID: &uid, Action: action, TargetType: "acp_session", TargetID: targetID, Metadata: meta,
	})
}