feat(app): wire ACP Supervisor + SessionService + startup reaper

internal/app/app.go

@@ -60,6 +60,7 @@ type App struct {
 	dispatcher *notify.Dispatcher
 	jobs       *jobs.Module
 	jobsCancel context.CancelFunc
+	acpSup     *acp.Supervisor // ACP supervisor — ShutdownAll on Run shutdown
 }
 
 // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误，不留半成品资源。
@@ -250,9 +251,85 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
 	chat.NewHandler(chatConvSvc, chatMsgSvc, chatTplSvc, chatEpSvc, chatUsageSvc,
 		chatUploader, userSvc, userAdminAdapter{svc: userSvc}).Mount(r)
 
-	// ===== ACP module wiring (Part 1: agent_kinds only) =====
+	// ===== ACP module wiring (Part 2: full SessionService + Supervisor) =====
 	acpRepo := acp.NewPostgresRepository(pool)
 	akSvc := acp.NewAgentKindService(acpRepo, enc, auditRec)
+
+	// Phase 1: startup reaper — recover sessions interrupted by previous crash.
+	// ResetStuckSessionsOnRestart marks all 'starting'/'running' sessions as 'crashed'
+	// and returns them; we release their worktrees and notify owners.
+	abortedSessions, rerr := acpRepo.ResetStuckSessionsOnRestart(ctx)
+	if rerr != nil {
+		log.Warn("acp.startup_reaper.reset_failed", "err", rerr.Error())
+	} else {
+		for _, sess := range abortedSessions {
+			if sess.IsMainWorktree {
+				if _, rerr := acpRepo.ReleaseMainWorktree(ctx, sess.WorkspaceID, sess.ID); rerr != nil {
+					log.Warn("acp.startup_reaper.release_main_failed",
+						"session_id", sess.ID, "err", rerr.Error())
+				}
+			} else {
+				if _, rerr := wtSvc.ReleaseByHolder(ctx, "session:"+sess.ID.String()); rerr != nil {
+					log.Warn("acp.startup_reaper.release_subwt_failed",
+						"session_id", sess.ID, "err", rerr.Error())
+				}
+			}
+			uid := sess.UserID
+			if rerr := auditRec.Record(ctx, audit.Entry{
+				UserID: &uid, Action: "acp.session.aborted_on_restart",
+				TargetType: "acp_session", TargetID: sess.ID.String(),
+				Metadata: map[string]any{
+					"agent_kind_id": sess.AgentKindID.String(),
+					"branch":        sess.Branch,
+				},
+			}); rerr != nil {
+				log.Warn("acp.startup_reaper.audit_failed",
+					"session_id", sess.ID, "err", rerr.Error())
+			}
+			if derr := notifyDispatcher.Dispatch(ctx, notify.Message{
+				UserID:   sess.UserID,
+				Topic:    "acp.session_crashed",
+				Severity: notify.SeverityWarning,
+				Title:    "ACP session terminated by server restart",
+				Body:     "Your session was interrupted while the server restarted.",
+				Metadata: map[string]any{
+					"session_id": sess.ID.String(),
+					"branch":     sess.Branch,
+				},
+			}); derr != nil {
+				log.Warn("acp.startup_reaper.notify_failed",
+					"session_id", sess.ID, "err", derr.Error())
+			}
+		}
+	}
+
+	// Phase 2: build supervisor + session service.
+	acpProjAccess := acpProjectAccessAdapter{wsRepo: wsRepo, projectRepo: projectRepo}
+	acpSup := acp.NewSupervisor(
+		acpRepo, auditRec, notifyDispatcher,
+		wtSvc, wsRepo, enc,
+		acp.SupervisorConfig{
+			SpawnTimeout:       cfg.Acp.SpawnTimeout,
+			KillGrace:          cfg.Acp.KillGrace,
+			StderrBufferLines:  cfg.Acp.StderrBufferLines,
+			StderrTailBytes:    cfg.Acp.StderrTailBytes,
+			EventMaxPayload:    cfg.Acp.EventMaxPayload,
+			EventTruncateField: cfg.Acp.EventTruncateField,
+			ShutdownGrace:      cfg.Acp.ShutdownGrace,
+		},
+		log,
+	)
+	sessSvc := acp.NewSessionService(
+		acpRepo, wsSvc, wtSvc, wsRepo,
+		acpProjAccess, acpSup, auditRec,
+		acp.SessionServiceConfig{
+			MaxActiveGlobal:  cfg.Acp.MaxActiveGlobal,
+			MaxActivePerUser: cfg.Acp.MaxActivePerUser,
+		},
+	)
+	_ = sessSvc // wired in G1 (handler refactor)
+
+	// Handler: G1 will refactor to take sessSvc + acpSup. For now use Part 1 signature.
 	acp.NewHandler(akSvc, userSvc, userAdminAdapter{svc: userSvc}, enc).Mount(r)
 
 	// ===== jobs module wiring =====
@@ -404,7 +481,13 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
 		WriteTimeout:      60 * time.Second,
 		IdleTimeout:       120 * time.Second,
 	}
-	return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher, jobs: jobsModule, jobsCancel: jobsCancel}, nil
+	return &App{
+		cfg: cfg, log: log, pool: pool, server: srv,
+		dispatcher: notifyDispatcher,
+		jobs:       jobsModule,
+		jobsCancel: jobsCancel,
+		acpSup:     acpSup,
+	}, nil
 }
 
 // Run 启动 HTTP server 并阻塞直到 ctx 取消（Shutdown）或服务器报错。
@@ -428,6 +511,11 @@ func (a *App) Run(ctx context.Context) error {
 		if shutErr != nil {
 			a.log.Error("server shutdown", "err", shutErr.Error())
 		}
+		// Stop ACP subprocesses (kill child agents) BEFORE jobs/dispatcher drain so
+		// any final crash notifications dispatched by onExit can flow through.
+		if a.acpSup != nil {
+			a.acpSup.ShutdownAll(shutCtx)
+		}
 		// Stop jobs module first so no in-flight worker can dispatch new
 		// notifications after we begin to drain the dispatcher.
 		if a.jobs != nil && a.cfg.Jobs.Enabled {
@@ -450,6 +538,9 @@ func (a *App) Run(ctx context.Context) error {
 	case err := <-errCh:
 		drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
+		if a.acpSup != nil {
+			a.acpSup.ShutdownAll(drainCtx)
+		}
 		if a.jobs != nil && a.cfg.Jobs.Enabled {
 			a.jobsCancel()
 			stopCtx, stopCancel := context.WithTimeout(context.Background(), a.cfg.Jobs.ShutdownGrace+5*time.Second)
@@ -524,6 +615,29 @@ func projectACL(pj *project.Project, callerID uuid.UUID, isAdmin bool) (canRead,
 	return canRead, canWrite
 }
 
+// acpProjectAccessAdapter satisfies acp.ProjectAccess. It uses workspace.Repository
+// to resolve workspace → project, then project.Repository for issue/requirement.
+type acpProjectAccessAdapter struct {
+	wsRepo      workspace.Repository
+	projectRepo project.Repository
+}
+
+func (a acpProjectAccessAdapter) GetProjectByWorkspace(ctx context.Context, wsID uuid.UUID) (*project.Project, error) {
+	ws, err := a.wsRepo.GetWorkspaceByID(ctx, wsID)
+	if err != nil {
+		return nil, err
+	}
+	return a.projectRepo.GetProjectByID(ctx, ws.ProjectID)
+}
+
+func (a acpProjectAccessAdapter) GetIssueByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Issue, error) {
+	return a.projectRepo.GetIssueByNumber(ctx, projectID, number)
+}
+
+func (a acpProjectAccessAdapter) GetRequirementByNumber(ctx context.Context, projectID uuid.UUID, number int) (*project.Requirement, error) {
+	return a.projectRepo.GetRequirementByNumber(ctx, projectID, number)
+}
+
 // ===== Adapters: bridge cross-module types into the narrow interfaces the
 // jobs/runners package expects, so the runners package never imports
 // workspace/chat/notify.