From 90f910fcdd8cdb0542f7e7172f304e2fed637e5d Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Thu, 7 May 2026 14:23:01 +0800
Subject: [PATCH] feat(acp): SessionService.Create with worktree acquire +
 supervisor.Spawn

---
 internal/acp/session_service.go | 236 ++++++++++++++++++++++++++++++--
 1 file changed, 228 insertions(+), 8 deletions(-)

diff --git a/internal/acp/session_service.go b/internal/acp/session_service.go
index fe12c01..f30e9c7 100644
--- a/internal/acp/session_service.go
+++ b/internal/acp/session_service.go
@@ -1,12 +1,13 @@
-// session_service.go 实现 SessionService 接口。Create 走完整事务（worktree
-// acquire + supervisor.Spawn）的实现见 F3；本文件仅放骨架 + ResolveBranch
-// pure helper。
+// session_service.go 实现 SessionService。Create 走完整事务（worktree
+// acquire + supervisor.Spawn）；Get/List/Terminate 是配套的简单读写。
 package acp
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 
@@ -98,19 +99,238 @@ func ResolveBranch(ctx context.Context, ws *workspace.Workspace, sessionID uuid.
 	return fmt.Sprintf("acp-auto/%s", short), false, nil
 }
 
-// Create / Get / List / Terminate 在 F3 实现；先放占位让接口断言通过。
+// 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) {
-	return nil, errs.New(errs.CodeInternal, "SessionService.Create not implemented (F3)")
+	// 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 创建）
+	var issueID, reqID *uuid.UUID
+	if in.IssueNumber != nil {
+		if iss, _ := s.pa.GetIssueByNumber(ctx, ws.ProjectID, *in.IssueNumber); iss != nil {
+			id := iss.ID
+			issueID = &id
+		}
+	}
+	if in.RequirementNumber != nil {
+		if req, _ := s.pa.GetRequirementByNumber(ctx, ws.ProjectID, *in.RequirementNumber); req != nil {
+			id := req.ID
+			reqID = &id
+		}
+	}
+
+	// 7. INSERT
+	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,
+	}
+	out, err := s.repo.InsertSession(ctx, sess)
+	if err != nil {
+		s.releaseOnFailure(ctx, sess, sessCaller, worktreeID)
+		return nil, err
+	}
+
+	// 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）
+	go func() {
+		spawnCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+		defer cancel()
+		if _, err := s.sup.Spawn(spawnCtx, out, kind); err != nil {
+			s.releaseOnFailure(spawnCtx, out, sessCaller, worktreeID)
+		}
+	}()
+
+	// 10. initial prompt：握手成功后异步转发
+	if in.InitialPrompt != nil && *in.InitialPrompt != "" {
+		go s.sendInitialPrompt(out.ID, *in.InitialPrompt)
+	}
+
+	return out, nil
+}
+
+// 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,
+		}
+		_ = relay.SendClient(msg)
+		return
+	}
 }
 
 func (s *sessionService) Get(ctx context.Context, c Caller, id uuid.UUID) (*Session, error) {
-	return nil, errs.New(errs.CodeInternal, "SessionService.Get not implemented (F3)")
+	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, all bool) ([]*Session, error) {
-	return nil, errs.New(errs.CodeInternal, "SessionService.List not implemented (F3)")
+	if all && !c.IsAdmin {
+		return nil, errs.New(errs.CodeForbidden, "admin only")
+	}
+	if all {
+		return s.repo.ListAllSessions(ctx)
+	}
+	return s.repo.ListSessionsByUser(ctx, c.UserID)
 }
 
 func (s *sessionService) Terminate(ctx context.Context, c Caller, id uuid.UUID) error {
-	return errs.New(errs.CodeInternal, "SessionService.Terminate not implemented (F3)")
+	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
+}
+
+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,
+	})
 }