From 26011975b499403dd34105b29d1bb8d28c59355f Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Thu, 7 May 2026 14:29:14 +0800 Subject: [PATCH] feat(app): wire ACP Supervisor + SessionService + startup reaper --- internal/app/app.go | 118 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 116 insertions(+), 2 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index cb8d989..6f80674 100644 --- a/internal/app/app.go +++ b/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.