From f4a7c770afe60e578fa4a8ef1d064ca7a9cd39cc Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 9 Jun 2026 22:43:29 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=AF=E5=8A=A8=E9=A1=B9=E7=9B=AE=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.example.yaml | 28 ++ internal/acp/sqlc/models.go | 18 + internal/acp/supervisor.go | 22 +- internal/acp/supervisor_unix.go | 23 - internal/acp/supervisor_windows.go | 14 - internal/app/app.go | 37 +- internal/audit/sqlc/models.go | 18 + internal/chat/sqlc/models.go | 18 + internal/config/config.go | 26 ++ internal/infra/errs/errs.go | 18 + internal/infra/notify/sqlc/models.go | 18 + internal/infra/proc/proc.go | 39 ++ internal/infra/proc/proc_test.go | 54 +++ internal/infra/proc/proc_unix.go | 50 +++ internal/infra/proc/proc_unix_test.go | 65 +++ internal/infra/proc/proc_windows.go | 73 +++ internal/jobs/sqlc/models.go | 18 + internal/mcp/sqlc/models.go | 18 + internal/project/sqlc/models.go | 18 + internal/run/domain.go | 41 ++ internal/run/dto.go | 121 +++++ internal/run/handler.go | 333 ++++++++++++++ internal/run/queries/run_profiles.sql | 45 ++ internal/run/relay.go | 214 +++++++++ internal/run/repository.go | 190 ++++++++ internal/run/run_profile_service.go | 425 ++++++++++++++++++ internal/run/sqlc/db.go | 32 ++ internal/run/sqlc/models.go | 345 ++++++++++++++ internal/run/sqlc/run_profiles.sql.go | 253 +++++++++++ internal/run/supervisor.go | 293 ++++++++++++ internal/run/supervisor_test.go | 163 +++++++ internal/user/sqlc/models.go | 18 + internal/workspace/sqlc/models.go | 18 + .../0010_workspace_run_profiles.down.sql | 1 + migrations/0010_workspace_run_profiles.up.sql | 23 + sqlc.yaml | 10 + web/src/api/runs.ts | 79 ++++ web/src/composables/useRunStream.ts | 128 ++++++ web/src/stores/runs.ts | 75 ++++ .../views/workspaces/WorkspaceHomeView.vue | 12 +- .../workspaces/components/RunLogViewer.vue | 79 ++++ .../components/RunProfileDialog.vue | 206 +++++++++ .../views/workspaces/components/RunsTab.vue | 160 +++++++ 43 files changed, 3797 insertions(+), 42 deletions(-) delete mode 100644 internal/acp/supervisor_unix.go delete mode 100644 internal/acp/supervisor_windows.go create mode 100644 internal/infra/proc/proc.go create mode 100644 internal/infra/proc/proc_test.go create mode 100644 internal/infra/proc/proc_unix.go create mode 100644 internal/infra/proc/proc_unix_test.go create mode 100644 internal/infra/proc/proc_windows.go create mode 100644 internal/run/domain.go create mode 100644 internal/run/dto.go create mode 100644 internal/run/handler.go create mode 100644 internal/run/queries/run_profiles.sql create mode 100644 internal/run/relay.go create mode 100644 internal/run/repository.go create mode 100644 internal/run/run_profile_service.go create mode 100644 internal/run/sqlc/db.go create mode 100644 internal/run/sqlc/models.go create mode 100644 internal/run/sqlc/run_profiles.sql.go create mode 100644 internal/run/supervisor.go create mode 100644 internal/run/supervisor_test.go create mode 100644 migrations/0010_workspace_run_profiles.down.sql create mode 100644 migrations/0010_workspace_run_profiles.up.sql create mode 100644 web/src/api/runs.ts create mode 100644 web/src/composables/useRunStream.ts create mode 100644 web/src/stores/runs.ts create mode 100644 web/src/views/workspaces/components/RunLogViewer.vue create mode 100644 web/src/views/workspaces/components/RunProfileDialog.vue create mode 100644 web/src/views/workspaces/components/RunsTab.vue diff --git a/config.example.yaml b/config.example.yaml index 1e5439a..4db4c21 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -147,3 +147,31 @@ mcp: rate_limit: per_token_per_minute: 100 global_per_minute: 1000 + +# Run profiles (后端托管 workspace 仓库启动) configuration +run: + enabled: true + kill_grace: 5s # stop 时优雅终止宽限，超时后整树 SIGKILL + out_buffer_lines: 1000 # 单次运行的合并 stdout+stderr 内存环形缓冲行数 + out_tail_bytes: 2000 # crash 时落 last_error 的尾部字节上限 + ws_send_buffer: 256 # 每个 WS 日志订阅者的发送缓冲 + shutdown_grace: 30s # 服务关停时整体终止所有 run 进程的总时限 + # 透传给被托管命令的宿主环境变量白名单。APP_MASTER_KEY / DB DSN / git 凭据 + # 等敏感变量绝不应加入此名单。 + env_whitelist: + - PATH + - HOME + - LANG + - LC_ALL + - TZ + - USER + - SHELL + - SystemRoot + - TEMP + - TMP + - USERPROFILE + - APPDATA + - LOCALAPPDATA + - PATHEXT + - ComSpec + - NUMBER_OF_PROCESSORS diff --git a/internal/acp/sqlc/models.go b/internal/acp/sqlc/models.go index 5c431d1..e8dcafb 100644 --- a/internal/acp/sqlc/models.go +++ b/internal/acp/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/acp/supervisor.go b/internal/acp/supervisor.go index 0dad597..6e76acc 100644 --- a/internal/acp/supervisor.go +++ b/internal/acp/supervisor.go @@ -27,6 +27,7 @@ import ( "github.com/yan1h/agent-coding-workflow/internal/audit" "github.com/yan1h/agent-coding-workflow/internal/infra/crypto" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" + procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc" "github.com/yan1h/agent-coding-workflow/internal/mcp" "github.com/yan1h/agent-coding-workflow/internal/workspace" ) @@ -95,6 +96,7 @@ type Process struct { WorktreeID *uuid.UUID // 子 worktree 时填，用于 release Cmd *exec.Cmd + group procgrp.Group // 进程树句柄（整树终止），Spawn 时 Prepare+Adopt Stdin io.WriteCloser Stdout io.ReadCloser Stderr io.ReadCloser @@ -173,6 +175,11 @@ func (s *Supervisor) Spawn(ctx context.Context, sess *Session, kind *AgentKind, cmd.Dir = sess.CwdPath cmd.Env = osEnv + // 进程树管理：Prepare 须在 Start 前（Unix Setpgid），Adopt 须在 Start 后 + // （Windows Job Object）。终止时整树清理，避免 agent 派生的子孙进程残留。 + group := procgrp.NewGroup() + group.Prepare(cmd) + stdin, err := cmd.StdinPipe() if err != nil { return nil, fmt.Errorf("stdin pipe: %w", err) @@ -189,6 +196,10 @@ func (s *Supervisor) Spawn(ctx context.Context, sess *Session, kind *AgentKind, if err := cmd.Start(); err != nil { return nil, fmt.Errorf("start: %w", err) } + if err := group.Adopt(cmd); err != nil { + _ = cmd.Process.Kill() + return nil, fmt.Errorf("adopt process group: %w", err) + } var decider handlers.PermissionDecider if s.permSvc != nil { @@ -215,6 +226,7 @@ func (s *Supervisor) Spawn(ctx context.Context, sess *Session, kind *AgentKind, UserID: sess.UserID, IsMain: sess.IsMainWorktree, Cmd: cmd, + group: group, Stdin: stdin, Stdout: stdout, Stderr: stderr, @@ -307,9 +319,9 @@ func (s *Supervisor) Kill(ctx context.Context, sid uuid.UUID, grace time.Duratio proc.KilledByUs.Store(true) - // 平台差异：Unix 走 SIGTERM + grace + 强 Kill；Windows 直接 Kill。 - // 实现见 supervisor_unix.go / supervisor_windows.go。 - s.unixGracefulTerm(proc, grace) + // 整树终止：Unix 走 SIGTERM(-pgid) + grace + SIGKILL(-pgid)；Windows 关闭 + // Job 句柄触发 KILL_ON_JOB_CLOSE。实现下沉至 internal/infra/proc。 + proc.group.TerminateGroup(proc.done, grace) select { case <-proc.done: @@ -375,6 +387,10 @@ func (s *Supervisor) monitorProcess(proc *Process) { delete(s.procs, proc.SessionID) s.mu.Unlock() + // 释放进程树句柄（Windows 关闭 Job handle；Unix no-op）。done 已 close、 + // 进程已 Wait 返回，此处 Close 不影响 onExit-must-not-relock 不变量。 + proc.group.Close() + status := SessionExited if !proc.KilledByUs.Load() { status = SessionCrashed diff --git a/internal/acp/supervisor_unix.go b/internal/acp/supervisor_unix.go deleted file mode 100644 index 6fb61d0..0000000 --- a/internal/acp/supervisor_unix.go +++ /dev/null @@ -1,23 +0,0 @@ -//go:build !windows - -package acp - -import ( - "syscall" - "time" -) - -// unixGracefulTerm 在 Unix 平台上先发 SIGTERM，等待 grace 时间内退出； -// 否则 Process.Kill。done channel 由 monitorProcess 关闭。 -func (s *Supervisor) unixGracefulTerm(proc *Process, grace time.Duration) { - if proc.Cmd.Process == nil { - return - } - _ = proc.Cmd.Process.Signal(syscall.SIGTERM) - select { - case <-proc.done: - return - case <-time.After(grace): - _ = proc.Cmd.Process.Kill() - } -} diff --git a/internal/acp/supervisor_windows.go b/internal/acp/supervisor_windows.go deleted file mode 100644 index 32445d1..0000000 --- a/internal/acp/supervisor_windows.go +++ /dev/null @@ -1,14 +0,0 @@ -//go:build windows - -package acp - -import "time" - -// unixGracefulTerm 在 Windows 上没有可靠的 SIGTERM 等价信号（CTRL_BREAK 仅 -// 对 console 子进程生效）。直接 Process.Kill，grace 参数不参与，仅签名一致。 -func (s *Supervisor) unixGracefulTerm(proc *Process, grace time.Duration) { - if proc.Cmd.Process == nil { - return - } - _ = proc.Cmd.Process.Kill() -} diff --git a/internal/app/app.go b/internal/app/app.go index f2917ad..1e277c3 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -46,6 +46,7 @@ import ( "github.com/yan1h/agent-coding-workflow/internal/jobs/runners" mcp "github.com/yan1h/agent-coding-workflow/internal/mcp" "github.com/yan1h/agent-coding-workflow/internal/project" + runmod "github.com/yan1h/agent-coding-workflow/internal/run" httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http" "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" "github.com/yan1h/agent-coding-workflow/internal/user" @@ -61,7 +62,8 @@ type App struct { dispatcher *notify.Dispatcher jobs *jobs.Module jobsCancel context.CancelFunc - acpSup *acp.Supervisor // ACP supervisor — ShutdownAll on Run shutdown + acpSup *acp.Supervisor // ACP supervisor — ShutdownAll on Run shutdown + runSup *runmod.RunSupervisor // run profiles supervisor — ShutdownAll on Run shutdown } // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误，不留半成品资源。 @@ -396,6 +398,31 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { SendBuffer: cfg.Acp.WSSendBuffer, }).Mount(r) + // ===== run profiles module wiring ===== + // 进程托管：每个 workspace 的 run profile 作为长驻进程在 main_path 拉起。 + // 复用 workspace.ProjectAccess(pa) 做 owner+admin 写鉴权。WS 心跳沿用 acp 的 + // 全局默认值（同类参数，避免新增配置面）。 + runRepo := runmod.NewPostgresRepository(pool) + runSup := runmod.NewSupervisor(runRepo, runmod.SupervisorConfig{ + KillGrace: cfg.Run.KillGrace, + OutBufferLines: cfg.Run.OutBufferLines, + OutTailBytes: cfg.Run.OutTailBytes, + WSSendBuffer: cfg.Run.WSSendBuffer, + ShutdownGrace: cfg.Run.ShutdownGrace, + }, log) + runSvc := runmod.NewService(runRepo, wsRepo, runSup, pa, auditRec, enc, runmod.ServiceConfig{ + EnvWhitelist: cfg.Run.EnvWhitelist, + KillGrace: cfg.Run.KillGrace, + OutTailBytes: cfg.Run.OutTailBytes, + }, log) + if cfg.Run.Enabled { + runmod.NewHandler(runSvc, userSvc, userAdminAdapter{svc: userSvc}, runmod.WSConfig{ + PingInterval: cfg.Acp.WSPingInterval, + PongTimeout: cfg.Acp.WSPongTimeout, + SendBuffer: cfg.Run.WSSendBuffer, + }).Mount(r) + } + // ===== jobs module wiring ===== jobsRepo := jobs.NewPostgresRepository(pool) @@ -577,6 +604,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { jobs: jobsModule, jobsCancel: jobsCancel, acpSup: acpSup, + runSup: runSup, }, nil } @@ -606,6 +634,10 @@ func (a *App) Run(ctx context.Context) error { if a.acpSup != nil { a.acpSup.ShutdownAll(shutCtx) } + // Stop run-profile subprocesses (整树 kill) on shutdown. + if a.runSup != nil { + a.runSup.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 { @@ -631,6 +663,9 @@ func (a *App) Run(ctx context.Context) error { if a.acpSup != nil { a.acpSup.ShutdownAll(drainCtx) } + if a.runSup != nil { + a.runSup.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) diff --git a/internal/audit/sqlc/models.go b/internal/audit/sqlc/models.go index 0854ff2..49e6200 100644 --- a/internal/audit/sqlc/models.go +++ b/internal/audit/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/chat/sqlc/models.go b/internal/chat/sqlc/models.go index 27c7979..93e1a58 100644 --- a/internal/chat/sqlc/models.go +++ b/internal/chat/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/config/config.go b/internal/config/config.go index ef4e203..4d2153e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -43,6 +43,7 @@ type Config struct { Notify NotifyConfig `mapstructure:"notify"` Acp AcpConfig `mapstructure:"acp"` MCP MCPConfig `mapstructure:"mcp"` + Run RunConfig `mapstructure:"run"` } // HTTPConfig 描述 HTTP 服务监听参数。 @@ -200,6 +201,20 @@ type AcpConfig struct { PermissionTimeout time.Duration `mapstructure:"permission_timeout"` } +// RunConfig 控制 workspace run profiles 模块：进程托管的终止宽限、日志环形缓冲 +// 容量、WS 发送缓冲、关停宽限，以及传递给被托管命令的宿主环境变量白名单。 +type RunConfig struct { + Enabled bool `mapstructure:"enabled"` + KillGrace time.Duration `mapstructure:"kill_grace"` + OutBufferLines int `mapstructure:"out_buffer_lines"` + OutTailBytes int `mapstructure:"out_tail_bytes"` + WSSendBuffer int `mapstructure:"ws_send_buffer"` + ShutdownGrace time.Duration `mapstructure:"shutdown_grace"` + // EnvWhitelist 是允许从宿主进程透传给被托管命令的环境变量名白名单。 + // APP_MASTER_KEY、DB DSN、git 凭据等敏感变量绝不应出现在此名单中。 + EnvWhitelist []string `mapstructure:"env_whitelist"` +} + // MCPConfig 控制 MCP 模块的开关、对外 URL（注入 ACP 子进程 env）、system token TTL 与速率限制。 type MCPConfig struct { Enabled bool `mapstructure:"enabled"` @@ -285,6 +300,17 @@ func Load(configFile string) (*Config, error) { v.SetDefault("mcp.system_token_ttl", "24h") v.SetDefault("mcp.rate_limit.per_token_per_minute", 100) v.SetDefault("mcp.rate_limit.global_per_minute", 1000) + v.SetDefault("run.enabled", true) + v.SetDefault("run.kill_grace", "5s") + v.SetDefault("run.out_buffer_lines", 1000) + v.SetDefault("run.out_tail_bytes", 2000) + v.SetDefault("run.ws_send_buffer", 256) + v.SetDefault("run.shutdown_grace", "30s") + v.SetDefault("run.env_whitelist", []string{ + "PATH", "HOME", "LANG", "LC_ALL", "TZ", "USER", "SHELL", + "SystemRoot", "TEMP", "TMP", "USERPROFILE", "APPDATA", "LOCALAPPDATA", + "PATHEXT", "ComSpec", "NUMBER_OF_PROCESSORS", + }) if configFile != "" { v.SetConfigFile(configFile) diff --git a/internal/infra/errs/errs.go b/internal/infra/errs/errs.go index 2a3229f..f37b2f1 100644 --- a/internal/infra/errs/errs.go +++ b/internal/infra/errs/errs.go @@ -99,6 +99,15 @@ const ( CodeMcpTokenNameRequired Code = "mcp.token_name_required" CodeMcpTokenExpiresRequired Code = "mcp.token_expires_required" CodeMcpRateLimited Code = "mcp.rate_limited" + + // Run 域（workspace run profiles） + CodeRunProfileNotFound Code = "run.profile_not_found" + CodeRunProfileSlugTaken Code = "run.profile_slug_taken" + CodeRunProfileAlreadyRunning Code = "run.profile_already_running" + CodeRunProfileNotRunning Code = "run.profile_not_running" + CodeRunProfileDisabled Code = "run.profile_disabled" + CodeRunSpawnFailed Code = "run.spawn_failed" + CodeRunCommandInvalid Code = "run.command_invalid" ) // AppError is the application's structured error type. It carries a stable @@ -239,6 +248,15 @@ func HTTPStatus(code Code) int { return http.StatusBadRequest case CodeMcpRateLimited: return http.StatusTooManyRequests + case CodeRunProfileNotFound: + return http.StatusNotFound + case CodeRunProfileSlugTaken, CodeRunProfileAlreadyRunning, + CodeRunProfileNotRunning, CodeRunProfileDisabled: + return http.StatusConflict + case CodeRunCommandInvalid: + return http.StatusBadRequest + case CodeRunSpawnFailed: + return http.StatusBadGateway default: return http.StatusInternalServerError } diff --git a/internal/infra/notify/sqlc/models.go b/internal/infra/notify/sqlc/models.go index 77645b3..471cef5 100644 --- a/internal/infra/notify/sqlc/models.go +++ b/internal/infra/notify/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/infra/proc/proc.go b/internal/infra/proc/proc.go new file mode 100644 index 0000000..abed1b8 --- /dev/null +++ b/internal/infra/proc/proc.go @@ -0,0 +1,39 @@ +// Package proc 提供跨平台的"进程组/进程树"终止抽象。一个 Group 把某个进程及其 +// 派生的所有子孙进程作为单一单元管理，使 stop/restart/crash 时能整树清理，避免 +// dev server（如 pnpm dev → node/vite/esbuild）留下占用端口的孤儿进程。 +// +// 用法与 *exec.Cmd 生命周期对齐： +// +// g := proc.NewGroup() +// g.Prepare(cmd) // cmd.Start() 之前：Unix 设置 SysProcAttr.Setpgid +// cmd.Start() +// g.Adopt(cmd) // cmd.Start() 之后：Windows 创建 Job 并 Assign +// ... +// g.TerminateGroup(done, grace) // 整树优雅终止 +// g.Close() // 释放平台句柄 +package proc + +import ( + "os/exec" + "time" +) + +// Group 抽象"一个进程及其所有子孙进程"作为单一单元管理。 +// 非并发安全：调用方需保证 Prepare/Adopt/TerminateGroup/Close 的串行使用 +// （RunSupervisor / ACP Supervisor 已用各自的锁与 done channel 保证）。 +type Group interface { + // Prepare 在 cmd.Start() 之前设置平台属性（Unix: SysProcAttr.Setpgid）。 + Prepare(cmd *exec.Cmd) + // Adopt 在 cmd.Start() 之后接管已启动的进程（Windows: 创建 Job 并 + // AssignProcessToJobObject）。Unix 为记录 pgid。必须在子进程派生前调用。 + Adopt(cmd *exec.Cmd) error + // TerminateGroup 优雅终止整组：Unix 先 SIGTERM(-pgid) 等 grace 再 SIGKILL(-pgid)； + // Windows CloseHandle(job) 触发 KILL_ON_JOB_CLOSE 整树终止。 + // done 由调用方（monitor goroutine）关闭，用于感知优雅退出、提前返回。 + TerminateGroup(done <-chan struct{}, grace time.Duration) + // Close 释放平台句柄（Windows: 关闭 Job handle；Unix: no-op）。幂等。 + Close() +} + +// NewGroup 按编译期 build tag 返回 unixGroup / windowsGroup。 +func NewGroup() Group { return newGroup() } diff --git a/internal/infra/proc/proc_test.go b/internal/infra/proc/proc_test.go new file mode 100644 index 0000000..8b59887 --- /dev/null +++ b/internal/infra/proc/proc_test.go @@ -0,0 +1,54 @@ +package proc + +import ( + "fmt" + "os/exec" + "runtime" + "testing" + "time" +) + +// sleeperCmd 返回一个会阻塞约 d 秒的跨平台命令。 +func sleeperCmd(d int) *exec.Cmd { + if runtime.GOOS == "windows" { + return exec.Command("cmd", "/c", fmt.Sprintf("timeout /t %d /nobreak >nul", d)) + } + return exec.Command("sh", "-c", fmt.Sprintf("sleep %d", d)) +} + +// TestGroup_TerminateKillsStartedProcess 验证完整生命周期：Prepare → Start → +// Adopt → TerminateGroup 能在 grace 内终止一个本应长跑的进程。 +func TestGroup_TerminateKillsStartedProcess(t *testing.T) { + g := NewGroup() + cmd := sleeperCmd(30) + g.Prepare(cmd) + if err := cmd.Start(); err != nil { + t.Fatalf("start: %v", err) + } + if err := g.Adopt(cmd); err != nil { + t.Fatalf("adopt: %v", err) + } + defer g.Close() + + done := make(chan struct{}) + go func() { + _ = cmd.Wait() + close(done) + }() + + g.TerminateGroup(done, 2*time.Second) + + select { + case <-done: + // 进程已终止，符合预期。 + case <-time.After(10 * time.Second): + t.Fatal("process not terminated within 10s after TerminateGroup") + } +} + +// TestNewGroup_NotNil 守护 build tag 选择逻辑：任意平台都应返回非 nil Group。 +func TestNewGroup_NotNil(t *testing.T) { + if NewGroup() == nil { + t.Fatal("NewGroup returned nil") + } +} diff --git a/internal/infra/proc/proc_unix.go b/internal/infra/proc/proc_unix.go new file mode 100644 index 0000000..00ac383 --- /dev/null +++ b/internal/infra/proc/proc_unix.go @@ -0,0 +1,50 @@ +//go:build !windows + +package proc + +import ( + "errors" + "os/exec" + "syscall" + "time" +) + +// unixGroup 用 POSIX 进程组实现整树终止：子进程经 Setpgid 成为新进程组组长 +// （pgid == pid），向负 pgid 发信号即覆盖整个进程组的所有子孙。 +type unixGroup struct { + pgid int +} + +func newGroup() Group { return &unixGroup{} } + +func (g *unixGroup) Prepare(cmd *exec.Cmd) { + if cmd.SysProcAttr == nil { + cmd.SysProcAttr = &syscall.SysProcAttr{} + } + cmd.SysProcAttr.Setpgid = true +} + +func (g *unixGroup) Adopt(cmd *exec.Cmd) error { + if cmd.Process == nil { + return errors.New("proc: Adopt before Start") + } + // Setpgid 使子进程成为新进程组组长，pgid == pid。 + g.pgid = cmd.Process.Pid + return nil +} + +func (g *unixGroup) TerminateGroup(done <-chan struct{}, grace time.Duration) { + if g.pgid <= 0 { + return + } + // 向负 pgid 发信号 = 向整个进程组（含所有子孙）发信号。忽略 ESRCH 等错误。 + _ = syscall.Kill(-g.pgid, syscall.SIGTERM) + select { + case <-done: + return + case <-time.After(grace): + _ = syscall.Kill(-g.pgid, syscall.SIGKILL) + } +} + +func (g *unixGroup) Close() {} diff --git a/internal/infra/proc/proc_unix_test.go b/internal/infra/proc/proc_unix_test.go new file mode 100644 index 0000000..ca3026e --- /dev/null +++ b/internal/infra/proc/proc_unix_test.go @@ -0,0 +1,65 @@ +//go:build !windows + +package proc + +import ( + "fmt" + "os/exec" + "syscall" + "testing" + "time" +) + +// TestUnixGroup_KillsChildProcess 验证整树清理：父 shell 派生一个后台 sleep +// 子进程并打印其 PID，TerminateGroup 后该子进程（孙级）也应被 kill(-pgid) 回收， +// 证明终止覆盖整个进程组而非仅组长。 +func TestUnixGroup_KillsChildProcess(t *testing.T) { + g := NewGroup() + // `sleep 30 &` 后台子进程；`echo $!` 打印其 PID；`wait` 让 shell 阻塞存活。 + cmd := exec.Command("sh", "-c", "sleep 30 & echo $!; wait") + g.Prepare(cmd) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("stdout pipe: %v", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("start: %v", err) + } + if err := g.Adopt(cmd); err != nil { + t.Fatalf("adopt: %v", err) + } + defer g.Close() + + var childPID int + if _, err := fmt.Fscan(stdout, &childPID); err != nil { + t.Fatalf("read child pid: %v", err) + } + if childPID <= 0 { + t.Fatalf("invalid child pid: %d", childPID) + } + + done := make(chan struct{}) + go func() { + _ = cmd.Wait() + close(done) + }() + + g.TerminateGroup(done, 2*time.Second) + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("parent not terminated") + } + + // 给内核一点时间回收子进程，然后用 kill(pid, 0) 探测：返回错误(ESRCH)=已死。 + deadline := time.Now().Add(5 * time.Second) + for { + if err := syscall.Kill(childPID, 0); err != nil { + return // 子进程已死，符合预期。 + } + if time.Now().After(deadline) { + t.Fatalf("child process %d still alive after group terminate", childPID) + } + time.Sleep(100 * time.Millisecond) + } +} diff --git a/internal/infra/proc/proc_windows.go b/internal/infra/proc/proc_windows.go new file mode 100644 index 0000000..2dcc2d3 --- /dev/null +++ b/internal/infra/proc/proc_windows.go @@ -0,0 +1,73 @@ +//go:build windows + +package proc + +import ( + "errors" + "os/exec" + "time" + "unsafe" + + "golang.org/x/sys/windows" +) + +// windowsGroup 用 Job Object 实现整树终止：把进程纳入一个设置了 +// KILL_ON_JOB_CLOSE 的 Job，此后该进程派生的所有子孙自动归入同一 Job， +// 关闭 Job 句柄即整树终止。优于 Process.Kill（只杀单 PID，漏 node 等子进程）。 +type windowsGroup struct { + job windows.Handle +} + +func newGroup() Group { return &windowsGroup{} } + +// Prepare 在 Windows 为 no-op：Job 须在进程 Start 之后才能 Assign。 +func (g *windowsGroup) Prepare(cmd *exec.Cmd) {} + +// Adopt 必须在 cmd.Start() 之后、子进程派生之前调用。竞态窗口：父进程 exec 后 +// 到首个子进程派生之间存在微秒级窗口（Go 无 pre-exec hook 无法完全消除）， +// 实践风险极小，文档化为已知限制。 +func (g *windowsGroup) Adopt(cmd *exec.Cmd) error { + if cmd.Process == nil { + return errors.New("proc: Adopt before Start") + } + job, err := windows.CreateJobObject(nil, nil) + if err != nil { + return err + } + var info windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION + info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if _, err := windows.SetInformationJobObject( + job, + windows.JobObjectExtendedLimitInformation, + uintptr(unsafe.Pointer(&info)), + uint32(unsafe.Sizeof(info)), + ); err != nil { + _ = windows.CloseHandle(job) + return err + } + h, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, false, uint32(cmd.Process.Pid)) + if err != nil { + _ = windows.CloseHandle(job) + return err + } + defer windows.CloseHandle(h) + if err := windows.AssignProcessToJobObject(job, h); err != nil { + _ = windows.CloseHandle(job) + return err + } + g.job = job + return nil +} + +// TerminateGroup 在 Windows 无可靠 SIGTERM 等价物，直接关闭 Job 句柄触发 +// KILL_ON_JOB_CLOSE 整树终止（grace/done 不参与，仅保持接口签名一致）。 +func (g *windowsGroup) TerminateGroup(_ <-chan struct{}, _ time.Duration) { + g.Close() +} + +func (g *windowsGroup) Close() { + if g.job != 0 { + _ = windows.CloseHandle(g.job) + g.job = 0 + } +} diff --git a/internal/jobs/sqlc/models.go b/internal/jobs/sqlc/models.go index 9aa07e0..0985142 100644 --- a/internal/jobs/sqlc/models.go +++ b/internal/jobs/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/mcp/sqlc/models.go b/internal/mcp/sqlc/models.go index 3d7ba86..6d58f23 100644 --- a/internal/mcp/sqlc/models.go +++ b/internal/mcp/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/project/sqlc/models.go b/internal/project/sqlc/models.go index 3d9b0c9..3e6cd48 100644 --- a/internal/project/sqlc/models.go +++ b/internal/project/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/run/domain.go b/internal/run/domain.go new file mode 100644 index 0000000..2569630 --- /dev/null +++ b/internal/run/domain.go @@ -0,0 +1,41 @@ +// Package run 实现 workspace run profiles：每个 workspace 下可配置多个运行档 +// （dev/prod/test），后端把每个 profile 作为长驻进程在 workspace 主工作区 +// （main_path）拉起并托管其完整生命周期（start/stop/restart + 实时日志）。 +// +// 运行态（running/stopped/crashed）只在 RunSupervisor 内存中维护，不落库； +// 仅最近一次运行的 last_started_at/last_exit_code/last_error 落行供历史回溯。 +package run + +import ( + "time" + + "github.com/google/uuid" +) + +// RunStatus 是 profile 的内存运行态（不持久化）。 +type RunStatus string + +const ( + StatusStopped RunStatus = "stopped" + StatusRunning RunStatus = "running" + StatusCrashed RunStatus = "crashed" +) + +// RunProfile 是一个 workspace 下的运行档（持久化实体）。 +type RunProfile struct { + ID uuid.UUID + WorkspaceID uuid.UUID + Slug string + Name string + Description string + Command string + Args []string + EncryptedEnv []byte + Enabled bool + LastStartedAt *time.Time + LastExitCode *int32 + LastError string + CreatedBy uuid.UUID + CreatedAt time.Time + UpdatedAt time.Time +} diff --git a/internal/run/dto.go b/internal/run/dto.go new file mode 100644 index 0000000..209f070 --- /dev/null +++ b/internal/run/dto.go @@ -0,0 +1,121 @@ +package run + +import ( + "sort" + "time" +) + +// ===== 请求 ===== + +// CreateRunProfileReq 是创建 run profile 的请求体。Env 为明文，服务端加密落库。 +type CreateRunProfileReq struct { + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + Env map[string]string `json:"env"` + Enabled *bool `json:"enabled"` // 省略默认为 true +} + +// UpdateRunProfileReq 是部分更新请求：nil 字段表示不修改。 +type UpdateRunProfileReq struct { + Slug *string `json:"slug"` + Name *string `json:"name"` + Description *string `json:"description"` + Command *string `json:"command"` + Args *[]string `json:"args"` + Env *map[string]string `json:"env"` + Enabled *bool `json:"enabled"` +} + +// ===== 响应 ===== + +// RunProfileResp 是 profile 的对外表示。env 不回传明文，仅回 env_keys。 +type RunProfileResp struct { + ID string `json:"id"` + WorkspaceID string `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EnvKeys []string `json:"env_keys"` + Enabled bool `json:"enabled"` + Status string `json:"status"` + StartedAt *string `json:"started_at"` + LastStartedAt *string `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +// RunStatusResp 是 start/stop/restart/status 的轻量返回。 +type RunStatusResp struct { + Status string `json:"status"` + StartedAt *string `json:"started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` +} + +// LogLineResp 是日志 tail 的返回行。 +type LogLineResp struct { + ID int64 `json:"id"` + Level string `json:"level"` + Text string `json:"text"` + TS string `json:"ts"` +} + +// ===== 转换 helper ===== + +func fmtTime(t time.Time) string { return t.UTC().Format(time.RFC3339) } + +func fmtTimePtr(t *time.Time) *string { + if t == nil { + return nil + } + s := t.UTC().Format(time.RFC3339) + return &s +} + +func buildRunProfileResp(p *RunProfile, envKeys []string, status RunStatus, startedAt *time.Time) RunProfileResp { + args := p.Args + if args == nil { + args = []string{} + } + if envKeys == nil { + envKeys = []string{} + } + return RunProfileResp{ + ID: p.ID.String(), + WorkspaceID: p.WorkspaceID.String(), + Slug: p.Slug, + Name: p.Name, + Description: p.Description, + Command: p.Command, + Args: args, + EnvKeys: envKeys, + Enabled: p.Enabled, + Status: string(status), + StartedAt: fmtTimePtr(startedAt), + LastStartedAt: fmtTimePtr(p.LastStartedAt), + LastExitCode: p.LastExitCode, + LastError: p.LastError, + CreatedAt: fmtTime(p.CreatedAt), + UpdatedAt: fmtTime(p.UpdatedAt), + } +} + +func sortedKeys(m map[string]string) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +func toLogLineResp(e *LogEnvelope) LogLineResp { + return LogLineResp{ID: e.ID, Level: e.Level, Text: e.Text, TS: e.TS.UTC().Format(time.RFC3339)} +} diff --git a/internal/run/handler.go b/internal/run/handler.go new file mode 100644 index 0000000..3b385e5 --- /dev/null +++ b/internal/run/handler.go @@ -0,0 +1,333 @@ +package run + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + + ws "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" + + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http" + "github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware" + "github.com/yan1h/agent-coding-workflow/internal/workspace" +) + +const defaultLogsLimit = 500 + +// AdminLookup resolves whether a user is an admin（mirrors acp/workspace pattern）。 +type AdminLookup interface { + IsAdmin(ctx context.Context, userID uuid.UUID) (bool, error) +} + +// WSConfig controls WebSocket heartbeat and buffering for the log stream. +type WSConfig struct { + PingInterval time.Duration + PongTimeout time.Duration + SendBuffer int +} + +// Handler 暴露 run profiles 的 HTTP/WS 端点。 +type Handler struct { + svc *Service + resolver middleware.SessionResolver + adminLookup AdminLookup + cfg WSConfig +} + +// NewHandler 构造 run.Handler。 +func NewHandler(svc *Service, resolver middleware.SessionResolver, al AdminLookup, cfg WSConfig) *Handler { + return &Handler{svc: svc, resolver: resolver, adminLookup: al, cfg: cfg} +} + +// Mount 注册 run profiles 路由。分两组：workspace 作用域（列表/创建）与 +// id 作用域（按 id 操作，内部经 profile→workspace 解析做 ACL）。 +func (h *Handler) Mount(r chi.Router) { + r.Route("/api/v1/workspaces/{wsID}/run-profiles", func(r chi.Router) { + r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{})) + r.Get("/", h.list) + r.Post("/", h.create) + }) + r.Route("/api/v1/run-profiles", func(r chi.Router) { + r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{})) + r.Patch("/{id}", h.update) + r.Delete("/{id}", h.remove) + r.Post("/{id}/start", h.start) + r.Post("/{id}/stop", h.stop) + r.Post("/{id}/restart", h.restart) + r.Get("/{id}/status", h.status) + r.Get("/{id}/logs", h.logs) + r.Get("/{id}/logs/stream", h.logsStream) + }) +} + +func (h *Handler) caller(r *http.Request) (workspace.Caller, error) { + uid, ok := middleware.UserIDFromContext(r.Context()) + if !ok { + return workspace.Caller{}, errs.New(errs.CodeUnauthorized, "not authenticated") + } + admin, err := h.adminLookup.IsAdmin(r.Context(), uid) + if err != nil { + return workspace.Caller{}, err + } + return workspace.Caller{UserID: uid, IsAdmin: admin}, nil +} + +func uuidParam(r *http.Request, key string) (uuid.UUID, error) { + id, err := uuid.Parse(chi.URLParam(r, key)) + if err != nil { + return uuid.Nil, errs.New(errs.CodeInvalidInput, "bad "+key) + } + return id, nil +} + +func (h *Handler) list(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + wsID, err := uuidParam(r, "wsID") + if err != nil { + writeErr(w, r, err) + return + } + out, err := h.svc.List(r.Context(), c, wsID) + if err != nil { + writeErr(w, r, err) + return + } + httpx.WriteJSON(w, http.StatusOK, out) +} + +func (h *Handler) create(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + wsID, err := uuidParam(r, "wsID") + if err != nil { + writeErr(w, r, err) + return + } + var req CreateRunProfileReq + if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil { + writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json")) + return + } + out, err := h.svc.Create(r.Context(), c, wsID, req) + if err != nil { + writeErr(w, r, err) + return + } + httpx.WriteJSON(w, http.StatusCreated, out) +} + +func (h *Handler) update(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + id, err := uuidParam(r, "id") + if err != nil { + writeErr(w, r, err) + return + } + var req UpdateRunProfileReq + if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil { + writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json")) + return + } + out, err := h.svc.Update(r.Context(), c, id, req) + if err != nil { + writeErr(w, r, err) + return + } + httpx.WriteJSON(w, http.StatusOK, out) +} + +func (h *Handler) remove(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + id, err := uuidParam(r, "id") + if err != nil { + writeErr(w, r, err) + return + } + if err := h.svc.Delete(r.Context(), c, id); err != nil { + writeErr(w, r, err) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func (h *Handler) start(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Start) } +func (h *Handler) stop(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Stop) } +func (h *Handler) restart(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Restart) } +func (h *Handler) status(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Status) } + +// lifecycle 复用 start/stop/restart/status 的公共骨架（鉴权 + 解析 id + 调 svc + 写 RunStatusResp）。 +func (h *Handler) lifecycle(w http.ResponseWriter, r *http.Request, + fn func(context.Context, workspace.Caller, uuid.UUID) (RunStatusResp, error)) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + id, err := uuidParam(r, "id") + if err != nil { + writeErr(w, r, err) + return + } + out, err := fn(r.Context(), c, id) + if err != nil { + writeErr(w, r, err) + return + } + httpx.WriteJSON(w, http.StatusOK, out) +} + +func (h *Handler) logs(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + id, err := uuidParam(r, "id") + if err != nil { + writeErr(w, r, err) + return + } + since := parseInt64(r.URL.Query().Get("since"), 0) + limit := int(parseInt64(r.URL.Query().Get("limit"), defaultLogsLimit)) + out, err := h.svc.Logs(r.Context(), c, id, since, limit) + if err != nil { + writeErr(w, r, err) + return + } + httpx.WriteJSON(w, http.StatusOK, out) +} + +// logsStream 升级为 WebSocket，推送内存日志（含 since 续传）与实时新行。只读流， +// 不接收客户端消息。鉴权经 middleware.Auth（WS 用 ?token= 传 token）。 +func (h *Handler) logsStream(w http.ResponseWriter, r *http.Request) { + c, err := h.caller(r) + if err != nil { + writeErr(w, r, err) + return + } + id, err := uuidParam(r, "id") + if err != nil { + writeErr(w, r, err) + return + } + since := parseInt64(r.URL.Query().Get("since"), 0) + + sub, relay, running, err := h.svc.SubscribeLogs(r.Context(), c, id, c.UserID) + if err != nil { + writeErr(w, r, err) + return + } + + conn, aerr := ws.Accept(w, r, &ws.AcceptOptions{}) + if aerr != nil { + return + } + defer conn.Close(ws.StatusInternalError, "internal") + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + // 初始 state 帧。 + status := StatusStopped + if running { + status = StatusRunning + } + if werr := wsjson.Write(ctx, conn, LogEnvelope{Kind: "state", Status: string(status), TS: time.Now()}); werr != nil { + return + } + + // profile 未运行：无 live relay，推完即关。 + if !running || relay == nil { + conn.Close(ws.StatusNormalClosure, "not running") + return + } + defer relay.Unsubscribe(sub.ID) + + // 推送历史缓冲（ID > since）。 + lastSent := since + for _, e := range relay.since(since, 0) { + if werr := wsjson.Write(ctx, conn, e); werr != nil { + return + } + lastSent = e.ID + } + + // 心跳。 + pingTicker := time.NewTicker(h.cfg.PingInterval) + defer pingTicker.Stop() + go func() { + for { + select { + case <-ctx.Done(): + return + case <-pingTicker.C: + pingCtx, pcancel := context.WithTimeout(ctx, h.cfg.PongTimeout) + perr := conn.Ping(pingCtx) + pcancel() + if perr != nil { + cancel() + return + } + } + } + }() + + for { + select { + case <-ctx.Done(): + return + case env, ok := <-sub.Send: + if !ok { + conn.Close(ws.StatusNormalClosure, "relay closed") + return + } + // 历史已推到 lastSent，避免重复推送同 ID 的日志行（控制帧 ID=0 放行）。 + if env.Kind == "log" && env.ID <= lastSent && env.ID != 0 { + continue + } + if werr := wsjson.Write(ctx, conn, env); werr != nil { + return + } + if env.ID > lastSent { + lastSent = env.ID + } + } + } +} + +func parseInt64(s string, def int64) int64 { + if s == "" { + return def + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return def + } + return n +} + +func writeErr(w http.ResponseWriter, r *http.Request, err error) { + httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err) +} diff --git a/internal/run/queries/run_profiles.sql b/internal/run/queries/run_profiles.sql new file mode 100644 index 0000000..d056ed9 --- /dev/null +++ b/internal/run/queries/run_profiles.sql @@ -0,0 +1,45 @@ +-- name: CreateRunProfile :one +INSERT INTO workspace_run_profiles ( + id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, created_by +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) +RETURNING *; + +-- name: GetRunProfileByID :one +SELECT * FROM workspace_run_profiles WHERE id = $1; + +-- name: ListRunProfilesByWorkspace :many +SELECT * FROM workspace_run_profiles +WHERE workspace_id = $1 +ORDER BY created_at; + +-- name: UpdateRunProfile :one +UPDATE workspace_run_profiles +SET slug = $2, + name = $3, + description = $4, + command = $5, + args = $6, + encrypted_env = $7, + enabled = $8, + updated_at = now() +WHERE id = $1 +RETURNING *; + +-- name: DeleteRunProfile :exec +DELETE FROM workspace_run_profiles WHERE id = $1; + +-- name: MarkRunProfileStarted :one +UPDATE workspace_run_profiles +SET last_started_at = now(), + last_exit_code = NULL, + last_error = '', + updated_at = now() +WHERE id = $1 +RETURNING *; + +-- name: UpdateRunProfileExit :exec +UPDATE workspace_run_profiles +SET last_exit_code = $2, + last_error = $3, + updated_at = now() +WHERE id = $1; diff --git a/internal/run/relay.go b/internal/run/relay.go new file mode 100644 index 0000000..5800b10 --- /dev/null +++ b/internal/run/relay.go @@ -0,0 +1,214 @@ +package run + +import ( + "log/slog" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" +) + +// LogEnvelope 是推到 WS / 经日志 tail 返回的一行日志或一个控制帧。 +// - kind="log"：普通日志行（level=info 来自 stdout，level=error 来自 stderr） +// - kind="state"：运行态变化（Status/ExitCode 有意义） +// - kind="slow_consumer_disconnect"：慢消费者被动断开的控制帧 +type LogEnvelope struct { + ID int64 `json:"id"` + Kind string `json:"kind"` + Level string `json:"level,omitempty"` + Text string `json:"text,omitempty"` + TS time.Time `json:"ts"` + Status string `json:"status,omitempty"` + ExitCode *int32 `json:"exit_code,omitempty"` +} + +// Subscriber 是单个 WS 连接的订阅句柄。 +type Subscriber struct { + ID uuid.UUID + UserID uuid.UUID + Send chan *LogEnvelope + closed atomic.Bool +} + +// outRing 是固定行数的环形缓冲：满则淘汰最旧行。承载某次运行的合并 stdout+stderr +// 日志，既供 logs(tail/since) 查询，也供 onExit 取尾部写 last_error。 +type outRing struct { + mu sync.Mutex + lines []*LogEnvelope + cap int +} + +func newOutRing(capacity int) *outRing { + if capacity <= 0 { + capacity = 1000 + } + return &outRing{cap: capacity} +} + +func (r *outRing) append(e *LogEnvelope) { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.lines) >= r.cap { + r.lines = r.lines[1:] + } + r.lines = append(r.lines, e) +} + +// since 返回 ID > sinceID 的日志行（最多 limit 条；limit<=0 表示不限）。 +func (r *outRing) since(sinceID int64, limit int) []*LogEnvelope { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*LogEnvelope, 0, len(r.lines)) + for _, e := range r.lines { + if e.ID > sinceID { + out = append(out, e) + } + } + if limit > 0 && len(out) > limit { + out = out[len(out)-limit:] + } + return out +} + +// tailText 反向重建尾部文本（不超过 maxBytes），供 onExit 写 last_error。 +func (r *outRing) tailText(maxBytes int) string { + r.mu.Lock() + defer r.mu.Unlock() + out := "" + for i := len(r.lines) - 1; i >= 0 && len(out) < maxBytes; i-- { + out = r.lines[i].Text + "\n" + out + } + if len(out) > maxBytes { + out = out[len(out)-maxBytes:] + } + return out +} + +// runRelay 是单次运行（一个 RunProc）的日志中继：合并 stdout+stderr 进环形 +// 缓冲并向所有 WS 订阅者 fanout。结构参照 acp/relay.go，但只做单向日志广播， +// 无 JSON-RPC 双向中继。 +type runRelay struct { + profileID uuid.UUID + ring *outRing + nextID atomic.Int64 + + subsMu sync.Mutex + subs map[uuid.UUID]*Subscriber + sendBuf int + + closed atomic.Bool + done chan struct{} + log *slog.Logger +} + +func newRunRelay(profileID uuid.UUID, bufLines, sendBuf int, log *slog.Logger) *runRelay { + if log == nil { + log = slog.Default() + } + if sendBuf <= 0 { + sendBuf = 256 + } + return &runRelay{ + profileID: profileID, + ring: newOutRing(bufLines), + subs: map[uuid.UUID]*Subscriber{}, + sendBuf: sendBuf, + done: make(chan struct{}), + log: log, + } +} + +// emit 记录一行日志（落环形缓冲 + fanout）。drainPipe goroutine 调用。 +func (r *runRelay) emit(level, text string) { + env := &LogEnvelope{ + ID: r.nextID.Add(1), + Kind: "log", + Level: level, + Text: text, + TS: time.Now(), + } + r.ring.append(env) + r.fanout(env) +} + +// Subscribe 注册一个 WS 订阅。 +func (r *runRelay) Subscribe(userID uuid.UUID) *Subscriber { + sub := &Subscriber{ + ID: uuid.New(), + UserID: userID, + Send: make(chan *LogEnvelope, r.sendBuf), + } + r.subsMu.Lock() + r.subs[sub.ID] = sub + r.subsMu.Unlock() + return sub +} + +// Unsubscribe 移除订阅；幂等。 +func (r *runRelay) Unsubscribe(subID uuid.UUID) { + r.subsMu.Lock() + defer r.subsMu.Unlock() + if sub, ok := r.subs[subID]; ok { + if sub.closed.CompareAndSwap(false, true) { + close(sub.Send) + } + delete(r.subs, subID) + } +} + +// fanout 把一条 envelope 推给所有订阅者；慢消费者（Send 满）自动断开， +// 不阻塞其他订阅者（同 acp relay 约定）。 +func (r *runRelay) fanout(env *LogEnvelope) { + r.subsMu.Lock() + defer r.subsMu.Unlock() + for id, sub := range r.subs { + select { + case sub.Send <- env: + default: + if sub.closed.CompareAndSwap(false, true) { + select { + case sub.Send <- &LogEnvelope{Kind: "slow_consumer_disconnect", TS: time.Now()}: + default: + } + close(sub.Send) + delete(r.subs, id) + r.log.Warn("run.relay.slow_subscriber_dropped", + "profile_id", r.profileID, "sub_id", id) + } + } + } +} + +// since 返回 ID > sinceID 的历史日志行（供 logs tail / WS 续传）。 +func (r *runRelay) since(sinceID int64, limit int) []*LogEnvelope { + return r.ring.since(sinceID, limit) +} + +// tailText 返回尾部文本（供 onExit 写 last_error）。 +func (r *runRelay) tailText(maxBytes int) string { + return r.ring.tailText(maxBytes) +} + +// Close 广播终止 state 帧后关闭所有订阅。幂等。 +func (r *runRelay) Close(status string, exitCode *int32) { + if !r.closed.CompareAndSwap(false, true) { + return + } + r.fanout(&LogEnvelope{ + ID: r.nextID.Add(1), + Kind: "state", + Status: status, + ExitCode: exitCode, + TS: time.Now(), + }) + r.subsMu.Lock() + for id, sub := range r.subs { + if sub.closed.CompareAndSwap(false, true) { + close(sub.Send) + } + delete(r.subs, id) + } + r.subsMu.Unlock() + close(r.done) +} diff --git a/internal/run/repository.go b/internal/run/repository.go new file mode 100644 index 0000000..1ea27df --- /dev/null +++ b/internal/run/repository.go @@ -0,0 +1,190 @@ +package run + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + runsqlc "github.com/yan1h/agent-coding-workflow/internal/run/sqlc" +) + +// Repository 是 run 模块对 PG 的全部依赖。Service / Supervisor 单测通过手写 +// fakeRepo 满足。 +type Repository interface { + CreateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) + GetRunProfileByID(ctx context.Context, id uuid.UUID) (*RunProfile, error) + ListRunProfilesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*RunProfile, error) + UpdateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) + DeleteRunProfile(ctx context.Context, id uuid.UUID) error + MarkRunProfileStarted(ctx context.Context, id uuid.UUID) (*RunProfile, error) + UpdateRunProfileExit(ctx context.Context, id uuid.UUID, exitCode *int32, lastErr string) error +} + +// IsUniqueViolation reports whether err is a PG unique constraint violation +// （service 层据此把 (workspace_id, slug) 冲突翻成 409）。 +func IsUniqueViolation(err error) bool { + var pg *pgconn.PgError + return errors.As(err, &pg) && pg.Code == pgerrcode.UniqueViolation +} + +type pgRepo struct { + q *runsqlc.Queries +} + +// NewPostgresRepository 用 pgxpool 构造 Repository。 +func NewPostgresRepository(pool *pgxpool.Pool) Repository { + return &pgRepo{q: runsqlc.New(pool)} +} + +// ===== 类型转换 helper ===== + +func toPgUUID(u uuid.UUID) pgtype.UUID { return pgtype.UUID{Bytes: u, Valid: true} } + +func fromPgUUID(p pgtype.UUID) uuid.UUID { + if !p.Valid { + return uuid.Nil + } + return uuid.UUID(p.Bytes) +} + +func fromPgTimePtr(t pgtype.Timestamptz) *time.Time { + if !t.Valid { + return nil + } + v := t.Time + return &v +} + +// normalizeStrSlice 把 nil 归一为非 nil 空切片，避免 pgx 把 nil []string 写成 +// SQL NULL 违反 TEXT[] NOT NULL 约束（同 acp/repository.go 的同名 helper）。 +func normalizeStrSlice(s []string) []string { + if s == nil { + return []string{} + } + return s +} + +func rowToRunProfile(row runsqlc.WorkspaceRunProfile) *RunProfile { + return &RunProfile{ + ID: fromPgUUID(row.ID), + WorkspaceID: fromPgUUID(row.WorkspaceID), + Slug: row.Slug, + Name: row.Name, + Description: row.Description, + Command: row.Command, + Args: row.Args, + EncryptedEnv: row.EncryptedEnv, + Enabled: row.Enabled, + LastStartedAt: fromPgTimePtr(row.LastStartedAt), + LastExitCode: row.LastExitCode, + LastError: row.LastError, + CreatedBy: fromPgUUID(row.CreatedBy), + CreatedAt: row.CreatedAt.Time, + UpdatedAt: row.UpdatedAt.Time, + } +} + +func (r *pgRepo) CreateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) { + row, err := r.q.CreateRunProfile(ctx, runsqlc.CreateRunProfileParams{ + ID: toPgUUID(p.ID), + WorkspaceID: toPgUUID(p.WorkspaceID), + Slug: p.Slug, + Name: p.Name, + Description: p.Description, + Command: p.Command, + Args: normalizeStrSlice(p.Args), + EncryptedEnv: p.EncryptedEnv, + Enabled: p.Enabled, + CreatedBy: toPgUUID(p.CreatedBy), + }) + if err != nil { + if IsUniqueViolation(err) { + return nil, errs.New(errs.CodeRunProfileSlugTaken, "run profile slug already taken in this workspace") + } + return nil, errs.Wrap(err, errs.CodeInternal, "create run profile") + } + return rowToRunProfile(row), nil +} + +func (r *pgRepo) GetRunProfileByID(ctx context.Context, id uuid.UUID) (*RunProfile, error) { + row, err := r.q.GetRunProfileByID(ctx, toPgUUID(id)) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found") + } + return nil, errs.Wrap(err, errs.CodeInternal, "get run profile") + } + return rowToRunProfile(row), nil +} + +func (r *pgRepo) ListRunProfilesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*RunProfile, error) { + rows, err := r.q.ListRunProfilesByWorkspace(ctx, toPgUUID(wsID)) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list run profiles") + } + out := make([]*RunProfile, 0, len(rows)) + for _, row := range rows { + out = append(out, rowToRunProfile(row)) + } + return out, nil +} + +func (r *pgRepo) UpdateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) { + row, err := r.q.UpdateRunProfile(ctx, runsqlc.UpdateRunProfileParams{ + ID: toPgUUID(p.ID), + Slug: p.Slug, + Name: p.Name, + Description: p.Description, + Command: p.Command, + Args: normalizeStrSlice(p.Args), + EncryptedEnv: p.EncryptedEnv, + Enabled: p.Enabled, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found") + } + if IsUniqueViolation(err) { + return nil, errs.New(errs.CodeRunProfileSlugTaken, "run profile slug already taken in this workspace") + } + return nil, errs.Wrap(err, errs.CodeInternal, "update run profile") + } + return rowToRunProfile(row), nil +} + +func (r *pgRepo) DeleteRunProfile(ctx context.Context, id uuid.UUID) error { + if err := r.q.DeleteRunProfile(ctx, toPgUUID(id)); err != nil { + return errs.Wrap(err, errs.CodeInternal, "delete run profile") + } + return nil +} + +func (r *pgRepo) MarkRunProfileStarted(ctx context.Context, id uuid.UUID) (*RunProfile, error) { + row, err := r.q.MarkRunProfileStarted(ctx, toPgUUID(id)) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found") + } + return nil, errs.Wrap(err, errs.CodeInternal, "mark run profile started") + } + return rowToRunProfile(row), nil +} + +func (r *pgRepo) UpdateRunProfileExit(ctx context.Context, id uuid.UUID, exitCode *int32, lastErr string) error { + if err := r.q.UpdateRunProfileExit(ctx, runsqlc.UpdateRunProfileExitParams{ + ID: toPgUUID(id), + LastExitCode: exitCode, + LastError: lastErr, + }); err != nil { + return errs.Wrap(err, errs.CodeInternal, "update run profile exit") + } + return nil +} diff --git a/internal/run/run_profile_service.go b/internal/run/run_profile_service.go new file mode 100644 index 0000000..cd63490 --- /dev/null +++ b/internal/run/run_profile_service.go @@ -0,0 +1,425 @@ +package run + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "os" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/yan1h/agent-coding-workflow/internal/audit" + "github.com/yan1h/agent-coding-workflow/internal/infra/crypto" + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + "github.com/yan1h/agent-coding-workflow/internal/workspace" +) + +// ServiceConfig 是 Service 的运行参数子集（派生自 config.RunConfig）。 +type ServiceConfig struct { + EnvWhitelist []string + KillGrace time.Duration + OutTailBytes int +} + +// Service 编排 run profile 的 CRUD 与进程生命周期（start/stop/restart/status/logs）。 +// 鉴权完全继承所属 workspace：读靠 visibility，写靠 owner+admin（经 ProjectAccess）。 +type Service struct { + repo Repository + wsRepo workspace.Repository + sup *RunSupervisor + pa workspace.ProjectAccess + rec audit.Recorder + enc *crypto.Encryptor + cfg ServiceConfig + log *slog.Logger +} + +// NewService 构造 run.Service。注入 workspace.Repository 以解析 main_path/project_id， +// ProjectAccess 做 owner/admin 写鉴权（复用 workspace 的窄接口）。 +func NewService( + repo Repository, + wsRepo workspace.Repository, + sup *RunSupervisor, + pa workspace.ProjectAccess, + rec audit.Recorder, + enc *crypto.Encryptor, + cfg ServiceConfig, + log *slog.Logger, +) *Service { + if log == nil { + log = slog.Default() + } + return &Service{repo: repo, wsRepo: wsRepo, sup: sup, pa: pa, rec: rec, enc: enc, cfg: cfg, log: log} +} + +// ===== 鉴权 helper ===== + +// authByWorkspace 解析 workspace 并校验调用方读/写权限。write=true 要求 owner+admin。 +func (s *Service) authByWorkspace(ctx context.Context, c workspace.Caller, wsID uuid.UUID, write bool) (*workspace.Workspace, error) { + ws, err := s.wsRepo.GetWorkspaceByID(ctx, wsID) + if err != nil { + return nil, err + } + canRead, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID) + if err != nil { + return nil, err + } + if write { + if !canWrite { + return nil, errs.New(errs.CodeForbidden, "no write access to this workspace") + } + } else if !canRead { + return nil, errs.New(errs.CodeForbidden, "no read access to this workspace") + } + return ws, nil +} + +// authByProfile 取 profile → 其 workspace，并校验权限。返回 profile 与 workspace。 +func (s *Service) authByProfile(ctx context.Context, c workspace.Caller, profileID uuid.UUID, write bool) (*RunProfile, *workspace.Workspace, error) { + p, err := s.repo.GetRunProfileByID(ctx, profileID) + if err != nil { + return nil, nil, err + } + ws, err := s.authByWorkspace(ctx, c, p.WorkspaceID, write) + if err != nil { + return nil, nil, err + } + return p, ws, nil +} + +// ===== CRUD ===== + +func (s *Service) List(ctx context.Context, c workspace.Caller, wsID uuid.UUID) ([]RunProfileResp, error) { + if _, err := s.authByWorkspace(ctx, c, wsID, false); err != nil { + return nil, err + } + profiles, err := s.repo.ListRunProfilesByWorkspace(ctx, wsID) + if err != nil { + return nil, err + } + out := make([]RunProfileResp, 0, len(profiles)) + for _, p := range profiles { + out = append(out, s.toResp(p)) + } + return out, nil +} + +func (s *Service) Create(ctx context.Context, c workspace.Caller, wsID uuid.UUID, req CreateRunProfileReq) (RunProfileResp, error) { + if _, err := s.authByWorkspace(ctx, c, wsID, true); err != nil { + return RunProfileResp{}, err + } + if strings.TrimSpace(req.Command) == "" { + return RunProfileResp{}, errs.New(errs.CodeRunCommandInvalid, "command must not be empty") + } + encEnv, err := s.encryptEnv(req.Env) + if err != nil { + return RunProfileResp{}, err + } + enabled := true + if req.Enabled != nil { + enabled = *req.Enabled + } + p := &RunProfile{ + ID: uuid.New(), + WorkspaceID: wsID, + Slug: req.Slug, + Name: req.Name, + Description: req.Description, + Command: req.Command, + Args: req.Args, + EncryptedEnv: encEnv, + Enabled: enabled, + CreatedBy: c.UserID, + } + created, err := s.repo.CreateRunProfile(ctx, p) + if err != nil { + return RunProfileResp{}, err + } + s.audit(ctx, c, "run.profile.created", created.ID, map[string]any{"workspace_id": wsID.String(), "slug": created.Slug}) + return s.toResp(created), nil +} + +func (s *Service) Update(ctx context.Context, c workspace.Caller, profileID uuid.UUID, req UpdateRunProfileReq) (RunProfileResp, error) { + p, _, err := s.authByProfile(ctx, c, profileID, true) + if err != nil { + return RunProfileResp{}, err + } + if req.Slug != nil { + p.Slug = *req.Slug + } + if req.Name != nil { + p.Name = *req.Name + } + if req.Description != nil { + p.Description = *req.Description + } + if req.Command != nil { + if strings.TrimSpace(*req.Command) == "" { + return RunProfileResp{}, errs.New(errs.CodeRunCommandInvalid, "command must not be empty") + } + p.Command = *req.Command + } + if req.Args != nil { + p.Args = *req.Args + } + if req.Enabled != nil { + p.Enabled = *req.Enabled + } + if req.Env != nil { + encEnv, encErr := s.encryptEnv(*req.Env) + if encErr != nil { + return RunProfileResp{}, encErr + } + p.EncryptedEnv = encEnv + } + updated, err := s.repo.UpdateRunProfile(ctx, p) + if err != nil { + return RunProfileResp{}, err + } + s.audit(ctx, c, "run.profile.updated", updated.ID, map[string]any{"workspace_id": updated.WorkspaceID.String()}) + return s.toResp(updated), nil +} + +func (s *Service) Delete(ctx context.Context, c workspace.Caller, profileID uuid.UUID) error { + p, _, err := s.authByProfile(ctx, c, profileID, true) + if err != nil { + return err + } + // 运行中先整树停掉，避免删行后留下无主进程。 + if _, running := s.sup.Get(profileID); running { + _ = s.sup.Stop(ctx, profileID, s.cfg.KillGrace) + } + if err := s.repo.DeleteRunProfile(ctx, profileID); err != nil { + return err + } + s.audit(ctx, c, "run.profile.deleted", profileID, map[string]any{"workspace_id": p.WorkspaceID.String()}) + return nil +} + +// ===== 进程生命周期 ===== + +func (s *Service) Start(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) { + p, ws, err := s.authByProfile(ctx, c, profileID, true) + if err != nil { + return RunStatusResp{}, err + } + if !p.Enabled { + return RunStatusResp{}, errs.New(errs.CodeRunProfileDisabled, "run profile is disabled") + } + overrides, err := s.decryptEnv(p.EncryptedEnv) + if err != nil { + return RunStatusResp{}, err + } + _, err = s.sup.Spawn(SpawnParams{ + ProfileID: p.ID, + Command: p.Command, + Args: p.Args, + Env: s.buildEnv(overrides), + Dir: ws.MainPath, + }) + if err != nil { + if errors.Is(err, ErrAlreadyRunning) { + return RunStatusResp{}, errs.New(errs.CodeRunProfileAlreadyRunning, "run profile already running") + } + return RunStatusResp{}, errs.Wrap(err, errs.CodeRunSpawnFailed, "spawn run profile") + } + if _, err := s.repo.MarkRunProfileStarted(ctx, p.ID); err != nil { + s.log.Error("run.start.mark_started", "profile_id", p.ID, "err", err.Error()) + } + s.audit(ctx, c, "run.profile.started", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()}) + return s.statusResp(p.ID), nil +} + +func (s *Service) Stop(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) { + p, _, err := s.authByProfile(ctx, c, profileID, true) + if err != nil { + return RunStatusResp{}, err + } + if err := s.sup.Stop(ctx, p.ID, s.cfg.KillGrace); err != nil { + if errors.Is(err, ErrNotRunning) { + return RunStatusResp{}, errs.New(errs.CodeRunProfileNotRunning, "run profile is not running") + } + return RunStatusResp{}, errs.Wrap(err, errs.CodeInternal, "stop run profile") + } + s.audit(ctx, c, "run.profile.stopped", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()}) + return s.statusResp(p.ID), nil +} + +func (s *Service) Restart(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) { + p, ws, err := s.authByProfile(ctx, c, profileID, true) + if err != nil { + return RunStatusResp{}, err + } + if !p.Enabled { + return RunStatusResp{}, errs.New(errs.CodeRunProfileDisabled, "run profile is disabled") + } + if _, running := s.sup.Get(p.ID); running { + _ = s.sup.Stop(ctx, p.ID, s.cfg.KillGrace) + } + overrides, err := s.decryptEnv(p.EncryptedEnv) + if err != nil { + return RunStatusResp{}, err + } + if _, err := s.sup.Spawn(SpawnParams{ + ProfileID: p.ID, + Command: p.Command, + Args: p.Args, + Env: s.buildEnv(overrides), + Dir: ws.MainPath, + }); err != nil { + if errors.Is(err, ErrAlreadyRunning) { + return RunStatusResp{}, errs.New(errs.CodeRunProfileAlreadyRunning, "run profile already running") + } + return RunStatusResp{}, errs.Wrap(err, errs.CodeRunSpawnFailed, "spawn run profile") + } + if _, err := s.repo.MarkRunProfileStarted(ctx, p.ID); err != nil { + s.log.Error("run.restart.mark_started", "profile_id", p.ID, "err", err.Error()) + } + s.audit(ctx, c, "run.profile.restarted", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()}) + return s.statusResp(p.ID), nil +} + +func (s *Service) Status(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) { + p, _, err := s.authByProfile(ctx, c, profileID, false) + if err != nil { + return RunStatusResp{}, err + } + return s.statusRespFrom(p), nil +} + +// Logs 返回正在运行进程的内存日志（ID > since，最多 limit 条）。 +func (s *Service) Logs(ctx context.Context, c workspace.Caller, profileID uuid.UUID, since int64, limit int) ([]LogLineResp, error) { + if _, _, err := s.authByProfile(ctx, c, profileID, false); err != nil { + return nil, err + } + envs := s.sup.LogsSince(profileID, since, limit) + out := make([]LogLineResp, 0, len(envs)) + for _, e := range envs { + out = append(out, toLogLineResp(e)) + } + return out, nil +} + +// SubscribeLogs 暴露给 WS handler：为运行中 profile 注册订阅，返回订阅句柄与 relay。 +// 同时返回鉴权后的 profile（未运行时 relay 为 nil）。 +func (s *Service) SubscribeLogs(ctx context.Context, c workspace.Caller, profileID, userID uuid.UUID) (*Subscriber, *runRelay, bool, error) { + if _, _, err := s.authByProfile(ctx, c, profileID, false); err != nil { + return nil, nil, false, err + } + sub, relay, ok := s.sup.Subscribe(profileID, userID) + return sub, relay, ok, nil +} + +// ===== 内部 helper ===== + +// buildEnv 构造被托管命令的环境：宿主白名单基础变量 + profile 覆盖。 +// 不继承宿主全部环境，避免泄漏 APP_MASTER_KEY / DB DSN / git 凭据等敏感变量。 +func (s *Service) buildEnv(overrides map[string]string) []string { + base := map[string]string{} + for _, k := range s.cfg.EnvWhitelist { + if v, ok := os.LookupEnv(k); ok { + base[k] = v + } + } + for k, v := range overrides { + base[k] = v + } + out := make([]string, 0, len(base)) + for k, v := range base { + out = append(out, k+"="+v) + } + return out +} + +func (s *Service) encryptEnv(env map[string]string) ([]byte, error) { + if len(env) == 0 { + return nil, nil + } + b, err := json.Marshal(env) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "marshal env") + } + ct, err := s.enc.Encrypt(b) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "encrypt env") + } + return ct, nil +} + +func (s *Service) decryptEnv(encrypted []byte) (map[string]string, error) { + if len(encrypted) == 0 { + return map[string]string{}, nil + } + plain, err := s.enc.Decrypt(encrypted) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "decrypt env") + } + var m map[string]string + if err := json.Unmarshal(plain, &m); err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "unmarshal env") + } + if m == nil { + m = map[string]string{} + } + return m, nil +} + +// envKeys 解密并返回排序后的 env key 列表（仅 key，供前端展示）。解密失败回空列表。 +func (s *Service) envKeys(encrypted []byte) []string { + m, err := s.decryptEnv(encrypted) + if err != nil { + s.log.Warn("run.env_keys.decrypt_failed", "err", err.Error()) + return []string{} + } + return sortedKeys(m) +} + +func (s *Service) toResp(p *RunProfile) RunProfileResp { + status := s.sup.StatusOf(p.ID) + var startedAt *time.Time + if proc, ok := s.sup.Get(p.ID); ok { + t := proc.StartedAt + startedAt = &t + } + return buildRunProfileResp(p, s.envKeys(p.EncryptedEnv), status, startedAt) +} + +func (s *Service) statusResp(profileID uuid.UUID) RunStatusResp { + p, err := s.repo.GetRunProfileByID(context.Background(), profileID) + if err != nil { + return RunStatusResp{Status: string(s.sup.StatusOf(profileID))} + } + return s.statusRespFrom(p) +} + +func (s *Service) statusRespFrom(p *RunProfile) RunStatusResp { + status := s.sup.StatusOf(p.ID) + var startedAt *string + if proc, ok := s.sup.Get(p.ID); ok { + t := proc.StartedAt.UTC().Format(time.RFC3339) + startedAt = &t + } + return RunStatusResp{ + Status: string(status), + StartedAt: startedAt, + LastExitCode: p.LastExitCode, + LastError: p.LastError, + } +} + +func (s *Service) audit(ctx context.Context, c workspace.Caller, action string, targetID uuid.UUID, meta map[string]any) { + if s.rec == nil { + return + } + uid := c.UserID + _ = s.rec.Record(ctx, audit.Entry{ + UserID: &uid, + Action: action, + TargetType: "run_profile", + TargetID: targetID.String(), + Metadata: meta, + }) +} diff --git a/internal/run/sqlc/db.go b/internal/run/sqlc/db.go new file mode 100644 index 0000000..6ed1a8a --- /dev/null +++ b/internal/run/sqlc/db.go @@ -0,0 +1,32 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package runsqlc + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type DBTX interface { + Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error) + Query(context.Context, string, ...interface{}) (pgx.Rows, error) + QueryRow(context.Context, string, ...interface{}) pgx.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx pgx.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/internal/run/sqlc/models.go b/internal/run/sqlc/models.go new file mode 100644 index 0000000..5371a07 --- /dev/null +++ b/internal/run/sqlc/models.go @@ -0,0 +1,345 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package runsqlc + +import ( + "net/netip" + + "github.com/jackc/pgx/v5/pgtype" +) + +type AcpAgentKind struct { + ID pgtype.UUID `json:"id"` + Name string `json:"name"` + DisplayName string `json:"display_name"` + Description string `json:"description"` + BinaryPath string `json:"binary_path"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + ToolAllowlist []string `json:"tool_allowlist"` +} + +type AcpEvent struct { + ID int64 `json:"id"` + SessionID pgtype.UUID `json:"session_id"` + Direction string `json:"direction"` + RpcKind string `json:"rpc_kind"` + Method *string `json:"method"` + Payload []byte `json:"payload"` + PayloadSize int32 `json:"payload_size"` + Truncated bool `json:"truncated"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type AcpPermissionRequest struct { + ID pgtype.UUID `json:"id"` + SessionID pgtype.UUID `json:"session_id"` + AgentRequestID string `json:"agent_request_id"` + ToolName string `json:"tool_name"` + ToolCall []byte `json:"tool_call"` + Options []byte `json:"options"` + Status string `json:"status"` + ChosenOptionID *string `json:"chosen_option_id"` + DecidedBy pgtype.UUID `json:"decided_by"` + DecidedAt pgtype.Timestamptz `json:"decided_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type AcpSession struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + ProjectID pgtype.UUID `json:"project_id"` + AgentKindID pgtype.UUID `json:"agent_kind_id"` + UserID pgtype.UUID `json:"user_id"` + IssueID pgtype.UUID `json:"issue_id"` + RequirementID pgtype.UUID `json:"requirement_id"` + AgentSessionID *string `json:"agent_session_id"` + Branch string `json:"branch"` + CwdPath string `json:"cwd_path"` + IsMainWorktree bool `json:"is_main_worktree"` + Status string `json:"status"` + Pid *int32 `json:"pid"` + ExitCode *int32 `json:"exit_code"` + LastError *string `json:"last_error"` + StartedAt pgtype.Timestamptz `json:"started_at"` + EndedAt pgtype.Timestamptz `json:"ended_at"` +} + +type AuditLog struct { + ID int64 `json:"id"` + UserID pgtype.UUID `json:"user_id"` + Action string `json:"action"` + TargetType *string `json:"target_type"` + TargetID *string `json:"target_id"` + Ip *netip.Addr `json:"ip"` + Metadata []byte `json:"metadata"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Conversation struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + ProjectID pgtype.UUID `json:"project_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + IssueID pgtype.UUID `json:"issue_id"` + ModelID pgtype.UUID `json:"model_id"` + SystemPrompt string `json:"system_prompt"` + Title *string `json:"title"` + TitleGeneratedAt pgtype.Timestamptz `json:"title_generated_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` +} + +type GitCredential struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Kind string `json:"kind"` + Username *string `json:"username"` + EncryptedSecret []byte `json:"encrypted_secret"` + Fingerprint *string `json:"fingerprint"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type Issue struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + RequirementID pgtype.UUID `json:"requirement_id"` + Number int32 `json:"number"` + Title string `json:"title"` + Description string `json:"description"` + Status string `json:"status"` + AssigneeID pgtype.UUID `json:"assignee_id"` + CreatedBy pgtype.UUID `json:"created_by"` + ClosedAt pgtype.Timestamptz `json:"closed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + WorkspaceID pgtype.UUID `json:"workspace_id"` +} + +type Job struct { + ID pgtype.UUID `json:"id"` + Type string `json:"type"` + Payload []byte `json:"payload"` + Status string `json:"status"` + ScheduledAt pgtype.Timestamptz `json:"scheduled_at"` + Attempts int32 `json:"attempts"` + MaxAttempts int32 `json:"max_attempts"` + LeasedAt pgtype.Timestamptz `json:"leased_at"` + LeasedBy *string `json:"leased_by"` + LastError *string `json:"last_error"` + CompletedAt pgtype.Timestamptz `json:"completed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmEndpoint struct { + ID pgtype.UUID `json:"id"` + Provider string `json:"provider"` + DisplayName string `json:"display_name"` + BaseUrl string `json:"base_url"` + ApiKeyEncrypted []byte `json:"api_key_encrypted"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmModel struct { + ID pgtype.UUID `json:"id"` + EndpointID pgtype.UUID `json:"endpoint_id"` + ModelID string `json:"model_id"` + DisplayName string `json:"display_name"` + Capabilities []byte `json:"capabilities"` + ContextWindow int32 `json:"context_window"` + MaxOutputTokens int32 `json:"max_output_tokens"` + PromptPricePerMillionUsd pgtype.Numeric `json:"prompt_price_per_million_usd"` + CompletionPricePerMillionUsd pgtype.Numeric `json:"completion_price_per_million_usd"` + ThinkingPricePerMillionUsd pgtype.Numeric `json:"thinking_price_per_million_usd"` + IsTitleGenerator bool `json:"is_title_generator"` + Enabled bool `json:"enabled"` + SortOrder int32 `json:"sort_order"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmUsage struct { + ID int64 `json:"id"` + ConversationID pgtype.UUID `json:"conversation_id"` + MessageID int64 `json:"message_id"` + UserID pgtype.UUID `json:"user_id"` + EndpointID pgtype.UUID `json:"endpoint_id"` + ModelID pgtype.UUID `json:"model_id"` + PromptTokens int32 `json:"prompt_tokens"` + CompletionTokens int32 `json:"completion_tokens"` + ThinkingTokens int32 `json:"thinking_tokens"` + CostUsd pgtype.Numeric `json:"cost_usd"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type McpToken struct { + ID pgtype.UUID `json:"id"` + TokenHash []byte `json:"token_hash"` + UserID pgtype.UUID `json:"user_id"` + Name string `json:"name"` + Issuer string `json:"issuer"` + Scope []byte `json:"scope"` + AcpSessionID pgtype.UUID `json:"acp_session_id"` + ExpiresAt pgtype.Timestamptz `json:"expires_at"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + RevokedAt pgtype.Timestamptz `json:"revoked_at"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Message struct { + ID int64 `json:"id"` + ConversationID pgtype.UUID `json:"conversation_id"` + Role string `json:"role"` + Content string `json:"content"` + Thinking *string `json:"thinking"` + ToolCalls []byte `json:"tool_calls"` + Status string `json:"status"` + ErrorMessage *string `json:"error_message"` + PromptTokens *int32 `json:"prompt_tokens"` + CompletionTokens *int32 `json:"completion_tokens"` + ThinkingTokens *int32 `json:"thinking_tokens"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type MessageAttachment struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + MessageID *int64 `json:"message_id"` + Filename string `json:"filename"` + MimeType string `json:"mime_type"` + SizeBytes int64 `json:"size_bytes"` + Sha256 string `json:"sha256"` + StoragePath string `json:"storage_path"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Notification struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + Topic string `json:"topic"` + Severity string `json:"severity"` + Title string `json:"title"` + Body string `json:"body"` + Link *string `json:"link"` + Metadata []byte `json:"metadata"` + ReadAt pgtype.Timestamptz `json:"read_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Project struct { + ID pgtype.UUID `json:"id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Visibility string `json:"visibility"` + OwnerID pgtype.UUID `json:"owner_id"` + ArchivedAt pgtype.Timestamptz `json:"archived_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type PromptTemplate struct { + ID pgtype.UUID `json:"id"` + Name string `json:"name"` + Content string `json:"content"` + Scope string `json:"scope"` + OwnerID pgtype.UUID `json:"owner_id"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type Requirement struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + Number int32 `json:"number"` + Title string `json:"title"` + Description string `json:"description"` + Phase string `json:"phase"` + Status string `json:"status"` + OwnerID pgtype.UUID `json:"owner_id"` + ClosedAt pgtype.Timestamptz `json:"closed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + WorkspaceID pgtype.UUID `json:"workspace_id"` +} + +type User struct { + ID pgtype.UUID `json:"id"` + Email string `json:"email"` + PasswordHash string `json:"password_hash"` + DisplayName string `json:"display_name"` + IsAdmin bool `json:"is_admin"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + Enabled bool `json:"enabled"` +} + +type UserSession struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + TokenHash []byte `json:"token_hash"` + ExpiresAt pgtype.Timestamptz `json:"expires_at"` + LastSeenAt pgtype.Timestamptz `json:"last_seen_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Workspace struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + GitRemoteUrl string `json:"git_remote_url"` + DefaultBranch string `json:"default_branch"` + MainPath string `json:"main_path"` + SyncStatus string `json:"sync_status"` + LastSyncedAt pgtype.Timestamptz `json:"last_synced_at"` + LastSyncError *string `json:"last_sync_error"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` +} + +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type WorkspaceWorktree struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Branch string `json:"branch"` + Path string `json:"path"` + Status string `json:"status"` + ActiveHolder *string `json:"active_holder"` + AcquiredAt pgtype.Timestamptz `json:"acquired_at"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + PruneWarningAt pgtype.Timestamptz `json:"prune_warning_at"` +} diff --git a/internal/run/sqlc/run_profiles.sql.go b/internal/run/sqlc/run_profiles.sql.go new file mode 100644 index 0000000..b706ac2 --- /dev/null +++ b/internal/run/sqlc/run_profiles.sql.go @@ -0,0 +1,253 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: run_profiles.sql + +package runsqlc + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const createRunProfile = `-- name: CreateRunProfile :one +INSERT INTO workspace_run_profiles ( + id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, created_by +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) +RETURNING id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, last_started_at, last_exit_code, last_error, created_by, created_at, updated_at +` + +type CreateRunProfileParams struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + CreatedBy pgtype.UUID `json:"created_by"` +} + +func (q *Queries) CreateRunProfile(ctx context.Context, arg CreateRunProfileParams) (WorkspaceRunProfile, error) { + row := q.db.QueryRow(ctx, createRunProfile, + arg.ID, + arg.WorkspaceID, + arg.Slug, + arg.Name, + arg.Description, + arg.Command, + arg.Args, + arg.EncryptedEnv, + arg.Enabled, + arg.CreatedBy, + ) + var i WorkspaceRunProfile + err := row.Scan( + &i.ID, + &i.WorkspaceID, + &i.Slug, + &i.Name, + &i.Description, + &i.Command, + &i.Args, + &i.EncryptedEnv, + &i.Enabled, + &i.LastStartedAt, + &i.LastExitCode, + &i.LastError, + &i.CreatedBy, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const deleteRunProfile = `-- name: DeleteRunProfile :exec +DELETE FROM workspace_run_profiles WHERE id = $1 +` + +func (q *Queries) DeleteRunProfile(ctx context.Context, id pgtype.UUID) error { + _, err := q.db.Exec(ctx, deleteRunProfile, id) + return err +} + +const getRunProfileByID = `-- name: GetRunProfileByID :one +SELECT id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, last_started_at, last_exit_code, last_error, created_by, created_at, updated_at FROM workspace_run_profiles WHERE id = $1 +` + +func (q *Queries) GetRunProfileByID(ctx context.Context, id pgtype.UUID) (WorkspaceRunProfile, error) { + row := q.db.QueryRow(ctx, getRunProfileByID, id) + var i WorkspaceRunProfile + err := row.Scan( + &i.ID, + &i.WorkspaceID, + &i.Slug, + &i.Name, + &i.Description, + &i.Command, + &i.Args, + &i.EncryptedEnv, + &i.Enabled, + &i.LastStartedAt, + &i.LastExitCode, + &i.LastError, + &i.CreatedBy, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const listRunProfilesByWorkspace = `-- name: ListRunProfilesByWorkspace :many +SELECT id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, last_started_at, last_exit_code, last_error, created_by, created_at, updated_at FROM workspace_run_profiles +WHERE workspace_id = $1 +ORDER BY created_at +` + +func (q *Queries) ListRunProfilesByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]WorkspaceRunProfile, error) { + rows, err := q.db.Query(ctx, listRunProfilesByWorkspace, workspaceID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []WorkspaceRunProfile + for rows.Next() { + var i WorkspaceRunProfile + if err := rows.Scan( + &i.ID, + &i.WorkspaceID, + &i.Slug, + &i.Name, + &i.Description, + &i.Command, + &i.Args, + &i.EncryptedEnv, + &i.Enabled, + &i.LastStartedAt, + &i.LastExitCode, + &i.LastError, + &i.CreatedBy, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const markRunProfileStarted = `-- name: MarkRunProfileStarted :one +UPDATE workspace_run_profiles +SET last_started_at = now(), + last_exit_code = NULL, + last_error = '', + updated_at = now() +WHERE id = $1 +RETURNING id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, last_started_at, last_exit_code, last_error, created_by, created_at, updated_at +` + +func (q *Queries) MarkRunProfileStarted(ctx context.Context, id pgtype.UUID) (WorkspaceRunProfile, error) { + row := q.db.QueryRow(ctx, markRunProfileStarted, id) + var i WorkspaceRunProfile + err := row.Scan( + &i.ID, + &i.WorkspaceID, + &i.Slug, + &i.Name, + &i.Description, + &i.Command, + &i.Args, + &i.EncryptedEnv, + &i.Enabled, + &i.LastStartedAt, + &i.LastExitCode, + &i.LastError, + &i.CreatedBy, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const updateRunProfile = `-- name: UpdateRunProfile :one +UPDATE workspace_run_profiles +SET slug = $2, + name = $3, + description = $4, + command = $5, + args = $6, + encrypted_env = $7, + enabled = $8, + updated_at = now() +WHERE id = $1 +RETURNING id, workspace_id, slug, name, description, command, args, encrypted_env, enabled, last_started_at, last_exit_code, last_error, created_by, created_at, updated_at +` + +type UpdateRunProfileParams struct { + ID pgtype.UUID `json:"id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` +} + +func (q *Queries) UpdateRunProfile(ctx context.Context, arg UpdateRunProfileParams) (WorkspaceRunProfile, error) { + row := q.db.QueryRow(ctx, updateRunProfile, + arg.ID, + arg.Slug, + arg.Name, + arg.Description, + arg.Command, + arg.Args, + arg.EncryptedEnv, + arg.Enabled, + ) + var i WorkspaceRunProfile + err := row.Scan( + &i.ID, + &i.WorkspaceID, + &i.Slug, + &i.Name, + &i.Description, + &i.Command, + &i.Args, + &i.EncryptedEnv, + &i.Enabled, + &i.LastStartedAt, + &i.LastExitCode, + &i.LastError, + &i.CreatedBy, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const updateRunProfileExit = `-- name: UpdateRunProfileExit :exec +UPDATE workspace_run_profiles +SET last_exit_code = $2, + last_error = $3, + updated_at = now() +WHERE id = $1 +` + +type UpdateRunProfileExitParams struct { + ID pgtype.UUID `json:"id"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` +} + +func (q *Queries) UpdateRunProfileExit(ctx context.Context, arg UpdateRunProfileExitParams) error { + _, err := q.db.Exec(ctx, updateRunProfileExit, arg.ID, arg.LastExitCode, arg.LastError) + return err +} diff --git a/internal/run/supervisor.go b/internal/run/supervisor.go new file mode 100644 index 0000000..9f395fb --- /dev/null +++ b/internal/run/supervisor.go @@ -0,0 +1,293 @@ +// supervisor.go 管理 run profile 子进程的内存运行态。每个 profile 同一时刻至多 +// 一个运行进程。并发骨架照搬 acp/supervisor.go 已被测试验证的不变量： +// - procs map 的写锁仅在 Spawn 插入 / monitor 删除时持有 +// - done channel 严格 close-once（monitorProcess 唯一持有） +// - KilledByUs atomic 区分 stopped（我方终止）vs crashed +// - onExit 不得重新 Lock s.mu（避免 monitor→Stop→onExit 链式死锁） +// +// 与 acp 的差异：无 JSON-RPC handshake / Relay 双向中继 / MCP token / worktree +// acquire；并补上经 internal/infra/proc 的整树终止（dev server 会派生子孙进程）。 +package run + +import ( + "bufio" + "context" + "io" + "log/slog" + "os/exec" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" + + procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc" +) + +// SupervisorConfig 是从 config.RunConfig 派生的 supervisor 子集。 +type SupervisorConfig struct { + KillGrace time.Duration + OutBufferLines int + OutTailBytes int + WSSendBuffer int + ShutdownGrace time.Duration +} + +// RunProc 是单个运行进程的运行时句柄。 +type RunProc struct { + ProfileID uuid.UUID + Cmd *exec.Cmd + group procgrp.Group + relay *runRelay + + StartedAt time.Time + KilledByUs atomic.Bool // true → stopped；false → crashed + done chan struct{} // monitor 退出时 close +} + +// RunSupervisor 是 run profile 子进程总管。被动结构：Spawn 时起 goroutine，无主循环。 +type RunSupervisor struct { + mu sync.RWMutex + procs map[uuid.UUID]*RunProc + terminal map[uuid.UUID]RunStatus // 最近一次退出的终止态（crashed/stopped），重启即清空 + + repo Repository + cfg SupervisorConfig + log *slog.Logger +} + +// NewSupervisor 构造空 RunSupervisor。 +func NewSupervisor(repo Repository, cfg SupervisorConfig, log *slog.Logger) *RunSupervisor { + if log == nil { + log = slog.Default() + } + return &RunSupervisor{ + procs: map[uuid.UUID]*RunProc{}, + terminal: map[uuid.UUID]RunStatus{}, + repo: repo, + cfg: cfg, + log: log, + } +} + +// SpawnParams 由 Service 构造：命令、参数、最终环境（宿主白名单+覆盖）、工作目录。 +type SpawnParams struct { + ProfileID uuid.UUID + Command string + Args []string + Env []string + Dir string +} + +// ErrAlreadyRunning / ErrNotRunning 是 supervisor 层的哨兵，Service 翻译为 errs code。 +var ( + ErrAlreadyRunning = errSentinel("run: profile already running") + ErrNotRunning = errSentinel("run: profile not running") +) + +type errSentinel string + +func (e errSentinel) Error() string { return string(e) } + +// Spawn 启动一个子进程并托管。调用方（Service）负责前置鉴权与 env 构造。 +// 同一 profile 已在运行时返回 ErrAlreadyRunning。 +func (s *RunSupervisor) Spawn(p SpawnParams) (*RunProc, error) { + s.mu.RLock() + _, running := s.procs[p.ProfileID] + s.mu.RUnlock() + if running { + return nil, ErrAlreadyRunning + } + + cmd := exec.Command(p.Command, p.Args...) + cmd.Dir = p.Dir + cmd.Env = p.Env + + // 进程树管理：Prepare 须在 Start 前（Unix Setpgid），Adopt 须在 Start 后 + // （Windows Job Object）。终止时整树清理，避免 dev server 子孙进程残留。 + group := procgrp.NewGroup() + group.Prepare(cmd) + + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, err + } + if err := cmd.Start(); err != nil { + return nil, err + } + if err := group.Adopt(cmd); err != nil { + _ = cmd.Process.Kill() + return nil, err + } + + proc := &RunProc{ + ProfileID: p.ProfileID, + Cmd: cmd, + group: group, + relay: newRunRelay(p.ProfileID, s.cfg.OutBufferLines, s.cfg.WSSendBuffer, s.log), + StartedAt: time.Now(), + done: make(chan struct{}), + } + + s.mu.Lock() + s.procs[p.ProfileID] = proc + delete(s.terminal, p.ProfileID) // 进入运行态，清除上一次终止态 + s.mu.Unlock() + + go s.drainPipe(proc, stdout, "info") + go s.drainPipe(proc, stderr, "error") + go s.monitorProcess(proc) + + return proc, nil +} + +// Stop 终止指定 profile 的子进程（整树）。幂等阻塞直到 done 关闭或 ctx 取消。 +// profile 未运行时返回 ErrNotRunning。 +func (s *RunSupervisor) Stop(ctx context.Context, profileID uuid.UUID, grace time.Duration) error { + s.mu.RLock() + proc, ok := s.procs[profileID] + s.mu.RUnlock() + if !ok { + return ErrNotRunning + } + proc.KilledByUs.Store(true) + proc.group.TerminateGroup(proc.done, grace) + select { + case <-proc.done: + case <-ctx.Done(): + } + return nil +} + +// Get 返回指定 profile 当前运行进程（用于 status/logs/subscribe）；未运行返回 false。 +func (s *RunSupervisor) Get(profileID uuid.UUID) (*RunProc, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + proc, ok := s.procs[profileID] + return proc, ok +} + +// StatusOf 返回内存运行态：在 procs 中 → running；否则取最近终止态（crashed/stopped）， +// 无记录（含服务端重启后）→ stopped。 +func (s *RunSupervisor) StatusOf(profileID uuid.UUID) RunStatus { + s.mu.RLock() + defer s.mu.RUnlock() + if _, ok := s.procs[profileID]; ok { + return StatusRunning + } + if st, ok := s.terminal[profileID]; ok { + return st + } + return StatusStopped +} + +// Subscribe 为正在运行的 profile 注册一个日志订阅；未运行返回 (nil,false)。 +func (s *RunSupervisor) Subscribe(profileID, userID uuid.UUID) (*Subscriber, *runRelay, bool) { + s.mu.RLock() + proc, ok := s.procs[profileID] + s.mu.RUnlock() + if !ok { + return nil, nil, false + } + return proc.relay.Subscribe(userID), proc.relay, true +} + +// LogsSince 返回正在运行进程的内存日志（ID > since）；未运行返回 nil。 +func (s *RunSupervisor) LogsSince(profileID uuid.UUID, since int64, limit int) []*LogEnvelope { + s.mu.RLock() + proc, ok := s.procs[profileID] + s.mu.RUnlock() + if !ok { + return nil + } + return proc.relay.since(since, limit) +} + +// drainPipe 逐行读取管道并 emit 到 relay。EOF/读错误时安静退出（Wait 由 monitor 处理）。 +func (s *RunSupervisor) drainPipe(proc *RunProc, pipe io.Reader, level string) { + scanner := bufio.NewScanner(pipe) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + proc.relay.emit(level, scanner.Text()) + } +} + +// monitorProcess 等待进程退出，更新 procs/terminal map，释放进程树句柄并触发 onExit。 +// 这是 done channel 唯一的合法关闭路径。 +func (s *RunSupervisor) monitorProcess(proc *RunProc) { + defer close(proc.done) + waitErr := proc.Cmd.Wait() + + exitCode := int32(0) + if exitErr, ok := waitErr.(*exec.ExitError); ok { + exitCode = int32(exitErr.ExitCode()) + } + + status := StatusStopped + if !proc.KilledByUs.Load() { + status = StatusCrashed + } + + s.mu.Lock() + delete(s.procs, proc.ProfileID) + s.terminal[proc.ProfileID] = status + s.mu.Unlock() + + // 释放进程树句柄（Windows 关闭 Job handle；Unix no-op）。 + proc.group.Close() + + s.onExit(context.Background(), proc, status, exitCode) +} + +// onExit 把退出信息落库并广播终止帧。不得再 Lock s.mu（monitorProcess 已 delete +// map 后调用本函数；Stop 也持读锁等 done，重入会死锁）。 +func (s *RunSupervisor) onExit(ctx context.Context, proc *RunProc, status RunStatus, exitCode int32) { + lastErr := "" + if status == StatusCrashed { + lastErr = proc.relay.tailText(s.cfg.OutTailBytes) + } + ec := exitCode + if s.repo != nil { + if err := s.repo.UpdateRunProfileExit(ctx, proc.ProfileID, &ec, lastErr); err != nil { + s.log.Error("run.on_exit.update_exit", "profile_id", proc.ProfileID, "err", err.Error()) + } + } + proc.relay.Close(string(status), &ec) +} + +// ShutdownAll 并行终止所有运行进程，受 cfg.ShutdownGrace 总时限约束。 +func (s *RunSupervisor) ShutdownAll(ctx context.Context) { + s.mu.RLock() + ids := make([]uuid.UUID, 0, len(s.procs)) + for id := range s.procs { + ids = append(ids, id) + } + s.mu.RUnlock() + + shutdownCtx, cancel := context.WithTimeout(ctx, s.cfg.ShutdownGrace) + defer cancel() + + var wg sync.WaitGroup + for _, id := range ids { + wg.Add(1) + go func(pid uuid.UUID) { + defer wg.Done() + _ = s.Stop(shutdownCtx, pid, s.cfg.KillGrace) + }(id) + } + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-shutdownCtx.Done(): + s.log.Warn("run.shutdown_all.timeout") + } +} diff --git a/internal/run/supervisor_test.go b/internal/run/supervisor_test.go new file mode 100644 index 0000000..54c0d18 --- /dev/null +++ b/internal/run/supervisor_test.go @@ -0,0 +1,163 @@ +package run + +import ( + "context" + "runtime" + "sync" + "testing" + "time" + + "github.com/google/uuid" +) + +// fakeRepo 仅实现 supervisor 实际用到的 UpdateRunProfileExit；其余满足接口即可。 +type fakeRepo struct { + mu sync.Mutex + exitID uuid.UUID + exitCode *int32 + lastErr string + called bool +} + +func (f *fakeRepo) CreateRunProfile(context.Context, *RunProfile) (*RunProfile, error) { + return nil, nil +} +func (f *fakeRepo) GetRunProfileByID(context.Context, uuid.UUID) (*RunProfile, error) { + return nil, nil +} +func (f *fakeRepo) ListRunProfilesByWorkspace(context.Context, uuid.UUID) ([]*RunProfile, error) { + return nil, nil +} +func (f *fakeRepo) UpdateRunProfile(context.Context, *RunProfile) (*RunProfile, error) { + return nil, nil +} +func (f *fakeRepo) DeleteRunProfile(context.Context, uuid.UUID) error { return nil } +func (f *fakeRepo) MarkRunProfileStarted(context.Context, uuid.UUID) (*RunProfile, error) { + return nil, nil +} +func (f *fakeRepo) UpdateRunProfileExit(_ context.Context, id uuid.UUID, exitCode *int32, lastErr string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.called = true + f.exitID = id + f.exitCode = exitCode + f.lastErr = lastErr + return nil +} + +func testCfg() SupervisorConfig { + return SupervisorConfig{ + KillGrace: 2 * time.Second, + OutBufferLines: 100, + OutTailBytes: 2000, + WSSendBuffer: 16, + ShutdownGrace: 5 * time.Second, + } +} + +func quickExit() SpawnParams { + p := SpawnParams{ProfileID: uuid.New()} + if runtime.GOOS == "windows" { + p.Command, p.Args = "cmd", []string{"/c", "exit", "0"} + } else { + p.Command, p.Args = "sh", []string{"-c", "exit 0"} + } + return p +} + +func sleeper() SpawnParams { + p := SpawnParams{ProfileID: uuid.New()} + if runtime.GOOS == "windows" { + p.Command, p.Args = "cmd", []string{"/c", "timeout", "/t", "30", "/nobreak"} + } else { + p.Command, p.Args = "sh", []string{"-c", "sleep 30"} + } + return p +} + +// TestSpawn_SelfExit_RecordsCrash 验证：未被我方终止而自行退出 → crashed， +// 且 onExit 把退出码落库（monitor 在 close(done) 前已完成 onExit）。 +func TestSpawn_SelfExit_RecordsCrash(t *testing.T) { + repo := &fakeRepo{} + sup := NewSupervisor(repo, testCfg(), nil) + params := quickExit() + + proc, err := sup.Spawn(params) + if err != nil { + t.Fatalf("spawn: %v", err) + } + select { + case <-proc.done: + case <-time.After(10 * time.Second): + t.Fatal("process did not exit") + } + + if got := sup.StatusOf(params.ProfileID); got != StatusCrashed { + t.Fatalf("status = %q, want crashed (self-exit not initiated by us)", got) + } + repo.mu.Lock() + defer repo.mu.Unlock() + if !repo.called { + t.Fatal("UpdateRunProfileExit not called on exit") + } + if repo.exitID != params.ProfileID { + t.Fatalf("exit recorded for wrong profile: %v", repo.exitID) + } + if repo.exitCode == nil || *repo.exitCode != 0 { + t.Fatalf("exit code = %v, want 0", repo.exitCode) + } +} + +// TestStop_TerminatesAndMarksStopped 验证 Stop 整树终止并判定为 stopped。 +func TestStop_TerminatesAndMarksStopped(t *testing.T) { + repo := &fakeRepo{} + sup := NewSupervisor(repo, testCfg(), nil) + params := sleeper() + + proc, err := sup.Spawn(params) + if err != nil { + t.Fatalf("spawn: %v", err) + } + if got := sup.StatusOf(params.ProfileID); got != StatusRunning { + t.Fatalf("status = %q, want running", got) + } + + if err := sup.Stop(context.Background(), params.ProfileID, testCfg().KillGrace); err != nil { + t.Fatalf("stop: %v", err) + } + // Stop 已等 done；onExit 在 close(done) 前完成。 + select { + case <-proc.done: + default: + t.Fatal("done not closed after Stop") + } + if got := sup.StatusOf(params.ProfileID); got != StatusStopped { + t.Fatalf("status = %q, want stopped", got) + } + if _, ok := sup.Get(params.ProfileID); ok { + t.Fatal("proc still registered after stop") + } +} + +// TestSpawn_AlreadyRunning 验证同一 profile 不能并发拉起两个进程。 +func TestSpawn_AlreadyRunning(t *testing.T) { + sup := NewSupervisor(&fakeRepo{}, testCfg(), nil) + params := sleeper() + + if _, err := sup.Spawn(params); err != nil { + t.Fatalf("first spawn: %v", err) + } + defer func() { _ = sup.Stop(context.Background(), params.ProfileID, testCfg().KillGrace) }() + + if _, err := sup.Spawn(params); err != ErrAlreadyRunning { + t.Fatalf("second spawn err = %v, want ErrAlreadyRunning", err) + } +} + +// TestStop_NotRunning 验证停止未运行的 profile 返回 ErrNotRunning。 +func TestStop_NotRunning(t *testing.T) { + sup := NewSupervisor(&fakeRepo{}, testCfg(), nil) + if err := sup.Stop(context.Background(), uuid.New(), testCfg().KillGrace); err != ErrNotRunning { + t.Fatalf("err = %v, want ErrNotRunning", err) + } +} diff --git a/internal/user/sqlc/models.go b/internal/user/sqlc/models.go index 182941d..c5babc0 100644 --- a/internal/user/sqlc/models.go +++ b/internal/user/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/internal/workspace/sqlc/models.go b/internal/workspace/sqlc/models.go index fa36ea3..c7c707e 100644 --- a/internal/workspace/sqlc/models.go +++ b/internal/workspace/sqlc/models.go @@ -313,6 +313,24 @@ type Workspace struct { ActiveMainSessionID pgtype.UUID `json:"active_main_session_id"` } +type WorkspaceRunProfile struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Command string `json:"command"` + Args []string `json:"args"` + EncryptedEnv []byte `json:"encrypted_env"` + Enabled bool `json:"enabled"` + LastStartedAt pgtype.Timestamptz `json:"last_started_at"` + LastExitCode *int32 `json:"last_exit_code"` + LastError string `json:"last_error"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + type WorkspaceWorktree struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/migrations/0010_workspace_run_profiles.down.sql b/migrations/0010_workspace_run_profiles.down.sql new file mode 100644 index 0000000..40d3899 --- /dev/null +++ b/migrations/0010_workspace_run_profiles.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS workspace_run_profiles; diff --git a/migrations/0010_workspace_run_profiles.up.sql b/migrations/0010_workspace_run_profiles.up.sql new file mode 100644 index 0000000..81d673a --- /dev/null +++ b/migrations/0010_workspace_run_profiles.up.sql @@ -0,0 +1,23 @@ +-- ============ Workspace Run Profiles ============ +-- 每个 workspace 下可配置多个运行档（dev/prod/test）。后端把每个 profile +-- 作为长驻进程拉起；运行态（running/stopped/crashed）在 RunSupervisor 内存中维护， +-- 不落库。下列三个 last_* 字段仅记录最近一次运行的历史，供前端展示与排障。 +CREATE TABLE workspace_run_profiles ( + id UUID PRIMARY KEY, + workspace_id UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + slug TEXT NOT NULL, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + command TEXT NOT NULL, -- 启动命令（exec 的 argv[0] 或 shell -c 串） + args TEXT[] NOT NULL DEFAULT '{}', -- 额外参数 + encrypted_env BYTEA, -- AES-GCM 密文（profile env 覆盖，map[string]string 的 JSON） + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_started_at TIMESTAMPTZ, -- 最近一次 start 时间 + last_exit_code INTEGER, -- 最近一次退出码（NULL=从未退出/从未运行） + last_error TEXT NOT NULL DEFAULT '', -- 最近一次 stderr 尾部 / 启动错误 + created_by UUID NOT NULL REFERENCES users(id) ON DELETE RESTRICT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (workspace_id, slug) +); +CREATE INDEX idx_run_profiles_workspace ON workspace_run_profiles(workspace_id); diff --git a/sqlc.yaml b/sqlc.yaml index ff2f45c..5964d11 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -90,3 +90,13 @@ sql: sql_package: pgx/v5 emit_pointers_for_null_types: true emit_json_tags: true + - engine: postgresql + queries: internal/run/queries + schema: migrations + gen: + go: + package: runsqlc + out: internal/run/sqlc + sql_package: pgx/v5 + emit_pointers_for_null_types: true + emit_json_tags: true diff --git a/web/src/api/runs.ts b/web/src/api/runs.ts new file mode 100644 index 0000000..dbaca61 --- /dev/null +++ b/web/src/api/runs.ts @@ -0,0 +1,79 @@ +import { request } from './client' + +export type RunState = 'running' | 'stopped' | 'crashed' + +export interface RunProfile { + id: string + workspace_id: string + slug: string + name: string + description: string + command: string + args: string[] + env_keys: string[] + enabled: boolean + status: RunState + started_at: string | null + last_started_at: string | null + last_exit_code: number | null + last_error: string + created_at: string + updated_at: string +} + +export interface RunStatus { + status: RunState + started_at: string | null + last_exit_code: number | null + last_error: string +} + +export interface LogLine { + id: number + level: 'info' | 'error' + text: string + ts: string +} + +export interface CreateRunProfileBody { + slug: string + name: string + description?: string + command: string + args?: string[] + env?: Record + enabled?: boolean +} + +export type UpdateRunProfileBody = Partial + +const enc = (v: string) => encodeURIComponent(v) + +export const runsApi = { + list: (wsID: string) => request(`/api/v1/workspaces/${enc(wsID)}/run-profiles/`), + create: (wsID: string, body: CreateRunProfileBody) => + request(`/api/v1/workspaces/${enc(wsID)}/run-profiles/`, { method: 'POST', body }), + update: (id: string, body: UpdateRunProfileBody) => + request(`/api/v1/run-profiles/${enc(id)}`, { method: 'PATCH', body }), + remove: (id: string) => request(`/api/v1/run-profiles/${enc(id)}`, { method: 'DELETE' }), + start: (id: string) => + request(`/api/v1/run-profiles/${enc(id)}/start`, { method: 'POST' }), + stop: (id: string) => + request(`/api/v1/run-profiles/${enc(id)}/stop`, { method: 'POST' }), + restart: (id: string) => + request(`/api/v1/run-profiles/${enc(id)}/restart`, { method: 'POST' }), + status: (id: string) => request(`/api/v1/run-profiles/${enc(id)}/status`), + logs: (id: string, since = 0) => + request(`/api/v1/run-profiles/${enc(id)}/logs?since=${since}`), +} + +// openRunLogsWS 打开到日志流的原生 WebSocket。浏览器 WS 不支持自定义 header， +// 故 token 经 ?token= 传递（同 acp WS 约定）。 +export function openRunLogsWS(profileID: string, since: number, token: string): WebSocket { + const proto = window.location.protocol === 'https:' ? 'wss' : 'ws' + const host = window.location.host + const url = + `${proto}://${host}/api/v1/run-profiles/${enc(profileID)}` + + `/logs/stream?since=${since}&token=${enc(token)}` + return new WebSocket(url) +} diff --git a/web/src/composables/useRunStream.ts b/web/src/composables/useRunStream.ts new file mode 100644 index 0000000..070216e --- /dev/null +++ b/web/src/composables/useRunStream.ts @@ -0,0 +1,128 @@ +// useRunStream 封装 run profile 的日志 WebSocket，结构照搬 useAcpStream： +// - connect + 自动重连（指数退避 1s/3s，最多 2 次） +// - ?since=lastLogID 续传 +// - 把日志行累积进 ref 数组；state 帧更新 runStatus +import { ref, onScopeDispose, type Ref } from 'vue' +import { openRunLogsWS, type LogLine, type RunState } from '@/api/runs' +import { useAuthStore } from '@/stores/auth' + +export type StreamStatus = 'idle' | 'connecting' | 'streaming' | 'closed' | 'error' + +export interface UseRunStreamReturn { + status: Ref + runStatus: Ref + logs: Ref + reconnect: () => void + close: () => void + clear: () => void +} + +interface WireFrame { + id?: number + kind?: string + level?: 'info' | 'error' + text?: string + ts?: string + status?: RunState +} + +export function useRunStream(profileId: string): UseRunStreamReturn { + const status = ref('idle') + const runStatus = ref('stopped') + const logs = ref([]) + const lastLogID = ref(0) + + let ws: WebSocket | null = null + let retried = 0 + let manualClose = false + let reconnectTimer: ReturnType | null = null + + const auth = useAuthStore() + + function connect() { + if (ws || manualClose) return + status.value = 'connecting' + const token = auth.token ?? '' + const sock = openRunLogsWS(profileId, lastLogID.value, token) + ws = sock + + sock.onopen = () => { + status.value = 'streaming' + retried = 0 + } + + sock.onmessage = (ev: MessageEvent) => { + let parsed: WireFrame + try { + parsed = JSON.parse(ev.data as string) as WireFrame + } catch { + return + } + if (!parsed || typeof parsed !== 'object') return + + if (parsed.kind === 'state') { + if (parsed.status) runStatus.value = parsed.status + return + } + if (parsed.kind === 'slow_consumer_disconnect') { + status.value = 'error' + return + } + // 普通日志行 + if (parsed.kind === 'log' && typeof parsed.id === 'number') { + if (parsed.id > lastLogID.value) lastLogID.value = parsed.id + logs.value.push({ + id: parsed.id, + level: parsed.level ?? 'info', + text: parsed.text ?? '', + ts: parsed.ts ?? '', + }) + } + } + + sock.onerror = () => { + status.value = 'error' + } + + sock.onclose = () => { + ws = null + if (manualClose) { + status.value = 'closed' + return + } + if (retried < 2) { + const delay = retried === 0 ? 1000 : 3000 + retried += 1 + reconnectTimer = setTimeout(connect, delay) + } else { + status.value = 'closed' + } + } + } + + function reconnect() { + if (ws) ws.close() + retried = 0 + manualClose = false + connect() + } + + function close() { + manualClose = true + if (reconnectTimer) { + clearTimeout(reconnectTimer) + reconnectTimer = null + } + if (ws) ws.close() + status.value = 'closed' + } + + function clear() { + logs.value = [] + } + + connect() + onScopeDispose(close) + + return { status, runStatus, logs, reconnect, close, clear } +} diff --git a/web/src/stores/runs.ts b/web/src/stores/runs.ts new file mode 100644 index 0000000..a7d8d32 --- /dev/null +++ b/web/src/stores/runs.ts @@ -0,0 +1,75 @@ +import { defineStore } from 'pinia' +import { ref } from 'vue' +import { + runsApi, + type RunProfile, + type RunStatus, + type CreateRunProfileBody, + type UpdateRunProfileBody, +} from '@/api/runs' + +export const useRunsStore = defineStore('runs', () => { + const profiles = ref([]) + const loading = ref(false) + + async function fetch(wsID: string) { + loading.value = true + try { + profiles.value = await runsApi.list(wsID) + } finally { + loading.value = false + } + } + + async function create(wsID: string, body: CreateRunProfileBody) { + const p = await runsApi.create(wsID, body) + profiles.value.unshift(p) + return p + } + + async function update(id: string, body: UpdateRunProfileBody) { + const p = await runsApi.update(id, body) + replace(p) + return p + } + + async function remove(id: string) { + await runsApi.remove(id) + profiles.value = profiles.value.filter((x) => x.id !== id) + } + + async function start(id: string) { + return applyStatus(id, await runsApi.start(id)) + } + async function stop(id: string) { + return applyStatus(id, await runsApi.stop(id)) + } + async function restart(id: string) { + return applyStatus(id, await runsApi.restart(id)) + } + async function refreshStatus(id: string) { + return applyStatus(id, await runsApi.status(id)) + } + + function replace(p: RunProfile) { + profiles.value = profiles.value.map((x) => (x.id === p.id ? p : x)) + } + + // applyStatus 用 start/stop/restart/status 返回的轻量状态更新对应行的运行态字段。 + function applyStatus(id: string, s: RunStatus): RunStatus { + profiles.value = profiles.value.map((x) => + x.id === id + ? { + ...x, + status: s.status, + started_at: s.started_at, + last_exit_code: s.last_exit_code, + last_error: s.last_error, + } + : x, + ) + return s + } + + return { profiles, loading, fetch, create, update, remove, start, stop, restart, refreshStatus } +}) diff --git a/web/src/views/workspaces/WorkspaceHomeView.vue b/web/src/views/workspaces/WorkspaceHomeView.vue index 4778e14..6416660 100644 --- a/web/src/views/workspaces/WorkspaceHomeView.vue +++ b/web/src/views/workspaces/WorkspaceHomeView.vue @@ -34,6 +34,15 @@ :ws="store.current" /> + + + ('worktrees') +const tab = ref<'worktrees' | 'main' | 'settings' | 'runs'>('worktrees') const projectSlug = computed(() => route.params.slug as string) const wsSlug = computed(() => route.params.wsSlug as string) diff --git a/web/src/views/workspaces/components/RunLogViewer.vue b/web/src/views/workspaces/components/RunLogViewer.vue new file mode 100644 index 0000000..481a550 --- /dev/null +++ b/web/src/views/workspaces/components/RunLogViewer.vue @@ -0,0 +1,79 @@ + + + diff --git a/web/src/views/workspaces/components/RunProfileDialog.vue b/web/src/views/workspaces/components/RunProfileDialog.vue new file mode 100644 index 0000000..5fca99d --- /dev/null +++ b/web/src/views/workspaces/components/RunProfileDialog.vue @@ -0,0 +1,206 @@ + + + diff --git a/web/src/views/workspaces/components/RunsTab.vue b/web/src/views/workspaces/components/RunsTab.vue new file mode 100644 index 0000000..6c81004 --- /dev/null +++ b/web/src/views/workspaces/components/RunsTab.vue @@ -0,0 +1,160 @@ + + +