|
| 1 | +// @index Shared MCP runtime assembly for stdio and HTTP server entry points. |
| 2 | +package mcpruntime |
| 3 | + |
| 4 | +import ( |
| 5 | + "context" |
| 6 | + "log/slog" |
| 7 | + "os/signal" |
| 8 | + "sync" |
| 9 | + "syscall" |
| 10 | + "time" |
| 11 | + |
| 12 | + mcpgo "github.com/mark3labs/mcp-go/server" |
| 13 | + "github.com/tae2089/trace" |
| 14 | + "gorm.io/gorm" |
| 15 | + |
| 16 | + "github.com/tae2089/code-context-graph/internal/analysis/changes" |
| 17 | + "github.com/tae2089/code-context-graph/internal/analysis/community" |
| 18 | + "github.com/tae2089/code-context-graph/internal/analysis/coupling" |
| 19 | + "github.com/tae2089/code-context-graph/internal/analysis/coverage" |
| 20 | + "github.com/tae2089/code-context-graph/internal/analysis/deadcode" |
| 21 | + "github.com/tae2089/code-context-graph/internal/analysis/flows" |
| 22 | + "github.com/tae2089/code-context-graph/internal/analysis/impact" |
| 23 | + "github.com/tae2089/code-context-graph/internal/analysis/largefunc" |
| 24 | + "github.com/tae2089/code-context-graph/internal/analysis/query" |
| 25 | + "github.com/tae2089/code-context-graph/internal/core" |
| 26 | + "github.com/tae2089/code-context-graph/internal/mcp" |
| 27 | + ccgobs "github.com/tae2089/code-context-graph/internal/obs" |
| 28 | + postprocesspolicy "github.com/tae2089/code-context-graph/internal/postprocess/policy" |
| 29 | +) |
| 30 | + |
| 31 | +// Options controls shared MCP runtime setup independent of transport. |
| 32 | +// @intent pass cache, telemetry, namespace, RAG, and parse-limit settings without importing HTTP server code. |
| 33 | +type Options struct { |
| 34 | + CacheTTL time.Duration |
| 35 | + NoCache bool |
| 36 | + OTELEndpoint string |
| 37 | + NamespaceRoot string |
| 38 | + WorkspaceRoot string |
| 39 | + RepoRoot string |
| 40 | + MaxFileBytes int64 |
| 41 | + MaxTotalParsedBytes int64 |
| 42 | + ServiceVersion string |
| 43 | + RagIndexDir string |
| 44 | + RagProjectDesc string |
| 45 | +} |
| 46 | + |
| 47 | +// Instance is a fully assembled MCP server plus runtime resources. |
| 48 | +// @intent share MCP server construction while keeping stdio and HTTP transports in separate packages. |
| 49 | +type Instance struct { |
| 50 | + Server *mcpgo.MCPServer |
| 51 | + Cache *mcp.Cache |
| 52 | + Deps *mcp.Deps |
| 53 | + PostprocessSummary func(context.Context) (*postprocesspolicy.StatusSummary, error) |
| 54 | + |
| 55 | + logger *slog.Logger |
| 56 | + shutdown func(context.Context) error |
| 57 | + close sync.Once |
| 58 | +} |
| 59 | + |
| 60 | +// New assembles MCP handlers, cache, telemetry, and postprocess policy. |
| 61 | +// @intent centralize common MCP dependency wiring without linking webhook/HTTP code into the local CLI binary. |
| 62 | +// @sideEffect initializes telemetry and optional in-memory cache. |
| 63 | +func New(rt *core.Runtime, opts Options) (*Instance, error) { |
| 64 | + rt.Logger.Info("starting code-context-graph MCP runtime") |
| 65 | + tel, err := ccgobs.Setup(context.Background(), ccgobs.Config{ |
| 66 | + ServiceName: "code-context-graph", |
| 67 | + ServiceVersion: opts.ServiceVersion, |
| 68 | + Mode: "serve", |
| 69 | + Endpoint: opts.OTELEndpoint, |
| 70 | + Logger: rt.Logger, |
| 71 | + }) |
| 72 | + if err != nil { |
| 73 | + return nil, trace.Wrap(err, "setup telemetry") |
| 74 | + } |
| 75 | + ccgobs.SetGlobal(tel) |
| 76 | + |
| 77 | + var cache *mcp.Cache |
| 78 | + if !opts.NoCache && opts.CacheTTL > 0 { |
| 79 | + cache = mcp.NewCache(opts.CacheTTL) |
| 80 | + rt.Logger.Info("MCP cache enabled", "ttl", opts.CacheTTL) |
| 81 | + } |
| 82 | + |
| 83 | + mcpWalkers := make(map[string]mcp.Parser, len(rt.Walkers)) |
| 84 | + for ext, w := range rt.Walkers { |
| 85 | + mcpWalkers[ext] = w |
| 86 | + } |
| 87 | + |
| 88 | + mcpDeps := &mcp.Deps{ |
| 89 | + Store: rt.Store, |
| 90 | + DB: rt.DB, |
| 91 | + Parser: rt.Walkers[".go"], |
| 92 | + Walkers: mcpWalkers, |
| 93 | + SearchBackend: rt.SearchBackend, |
| 94 | + ImpactAnalyzer: impact.New(rt.Store), |
| 95 | + FlowTracer: flows.New(rt.Store), |
| 96 | + ChangesGitClient: changes.NewExecGitClient(), |
| 97 | + QueryService: query.New(rt.DB), |
| 98 | + LargefuncAnalyzer: largefunc.New(rt.DB), |
| 99 | + DeadcodeAnalyzer: deadcode.New(rt.DB), |
| 100 | + CouplingAnalyzer: coupling.New(rt.DB), |
| 101 | + CoverageAnalyzer: coverage.New(rt.DB), |
| 102 | + CommunityBuilder: community.New(rt.DB), |
| 103 | + FlowBuilder: flows.NewBuilder(rt.DB, rt.Store), |
| 104 | + Incremental: rt.Syncer, |
| 105 | + PostprocessPolicy: NewPostprocessPolicy(rt.DB), |
| 106 | + Logger: rt.Logger, |
| 107 | + Cache: cache, |
| 108 | + RagIndexDir: opts.RagIndexDir, |
| 109 | + RagProjectDesc: opts.RagProjectDesc, |
| 110 | + NamespaceRoot: opts.NamespaceRoot, |
| 111 | + WorkspaceRoot: opts.WorkspaceRoot, |
| 112 | + RepoRoot: opts.RepoRoot, |
| 113 | + MaxFileBytes: opts.MaxFileBytes, |
| 114 | + MaxTotalParsedBytes: opts.MaxTotalParsedBytes, |
| 115 | + } |
| 116 | + |
| 117 | + inst := &Instance{ |
| 118 | + Server: mcp.NewServer(mcpDeps), |
| 119 | + Cache: cache, |
| 120 | + Deps: mcpDeps, |
| 121 | + logger: rt.Logger, |
| 122 | + shutdown: tel.Shutdown, |
| 123 | + } |
| 124 | + inst.PostprocessSummary = func(ctx context.Context) (*postprocesspolicy.StatusSummary, error) { |
| 125 | + if mcpDeps.PostprocessPolicy == nil { |
| 126 | + return nil, nil |
| 127 | + } |
| 128 | + return mcpDeps.PostprocessPolicy.Status(ctx, postprocesspolicy.StatusOptions{RecentLimit: postprocesspolicy.DefaultStatusLimit}) |
| 129 | + } |
| 130 | + return inst, nil |
| 131 | +} |
| 132 | + |
| 133 | +// Close releases MCP cache and telemetry resources. |
| 134 | +// @intent provide one idempotent cleanup path for transport-specific runners. |
| 135 | +// @sideEffect closes the cache and shuts down telemetry with a bounded context. |
| 136 | +func (i *Instance) Close() { |
| 137 | + i.close.Do(func() { |
| 138 | + if i.Cache != nil { |
| 139 | + i.Cache.Close() |
| 140 | + } |
| 141 | + if i.shutdown != nil { |
| 142 | + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 143 | + defer cancel() |
| 144 | + if err := i.shutdown(shutdownCtx); err != nil { |
| 145 | + i.logger.Error("telemetry shutdown failed", "error", err) |
| 146 | + } |
| 147 | + } |
| 148 | + ccgobs.SetGlobal(nil) |
| 149 | + }) |
| 150 | +} |
| 151 | + |
| 152 | +// RunStdio assembles and serves MCP over stdio. |
| 153 | +// @intent keep the local ccg binary on a stdio-only runtime without importing HTTP/webhook server code. |
| 154 | +// @sideEffect starts a long-running stdio MCP process and handles SIGINT/SIGTERM shutdown. |
| 155 | +func RunStdio(rt *core.Runtime, opts Options) error { |
| 156 | + inst, err := New(rt, opts) |
| 157 | + if err != nil { |
| 158 | + return err |
| 159 | + } |
| 160 | + defer inst.Close() |
| 161 | + |
| 162 | + rt.Logger.Info("serving MCP over stdio") |
| 163 | + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) |
| 164 | + defer stop() |
| 165 | + errCh := make(chan error, 1) |
| 166 | + go func() { |
| 167 | + errCh <- mcpgo.ServeStdio(inst.Server) |
| 168 | + }() |
| 169 | + select { |
| 170 | + case err := <-errCh: |
| 171 | + if err != nil { |
| 172 | + return trace.Wrap(err, "MCP server") |
| 173 | + } |
| 174 | + case <-ctx.Done(): |
| 175 | + rt.Logger.Info("received signal, shutting down stdio MCP server") |
| 176 | + } |
| 177 | + return nil |
| 178 | +} |
| 179 | + |
| 180 | +// FlushQueryCache clears the MCP query cache if it exists. |
| 181 | +// @intent let graph updates invalidate shared MCP cache without coupling to transport packages. |
| 182 | +// @sideEffect cache가 있으면 저장된 질의 결과를 비운다. |
| 183 | +func FlushQueryCache(cache *mcp.Cache) { |
| 184 | + if cache != nil { |
| 185 | + cache.Flush() |
| 186 | + } |
| 187 | +} |
| 188 | + |
| 189 | +// MCPPostprocessPolicy manages post-processing policies for the MCP runtime. |
| 190 | +// @intent MCP 실행 경로가 후처리 정책 결정을 공통 래퍼로 호출하게 한다. |
| 191 | +type MCPPostprocessPolicy struct { |
| 192 | + engine *postprocesspolicy.Engine |
| 193 | + store *postprocesspolicy.Store |
| 194 | +} |
| 195 | + |
| 196 | +// NewPostprocessPolicy creates a new MCP post-processing policy wrapper. |
| 197 | +// @intent MCP 실행 경로에서 후처리 정책 엔진과 저장소를 함께 묶어 제공한다. |
| 198 | +func NewPostprocessPolicy(db *gorm.DB) *MCPPostprocessPolicy { |
| 199 | + if db == nil { |
| 200 | + return nil |
| 201 | + } |
| 202 | + return &MCPPostprocessPolicy{ |
| 203 | + engine: &postprocesspolicy.Engine{}, |
| 204 | + store: postprocesspolicy.NewStore(db), |
| 205 | + } |
| 206 | +} |
| 207 | + |
| 208 | +// Resolve decides the policy for a given tool and input. |
| 209 | +// @intent 요청된 후처리 도구에 적용할 정책과 출처를 계산한다. |
| 210 | +func (p *MCPPostprocessPolicy) Resolve(ctx context.Context, input postprocesspolicy.DecisionInput) (string, string, error) { |
| 211 | + return p.engine.Resolve(ctx, p.store, input) |
| 212 | +} |
| 213 | + |
| 214 | +// RecordRun logs the results of a post-processing run. |
| 215 | +// @intent 후처리 실행 결과를 정책 저장소에 기록해 후속 판단에 반영한다. |
| 216 | +// @sideEffect ccg_postprocess_run_logs 상태를 갱신한다. |
| 217 | +func (p *MCPPostprocessPolicy) RecordRun(ctx context.Context, record postprocesspolicy.RunRecord) error { |
| 218 | + return p.store.RecordRun(ctx, record) |
| 219 | +} |
| 220 | + |
| 221 | +// Status returns the current status summary of post-processing. |
| 222 | +// @intent 운영 상태 엔드포인트가 후처리 건강 상태를 요약해서 볼 수 있게 한다. |
| 223 | +func (p *MCPPostprocessPolicy) Status(ctx context.Context, opts postprocesspolicy.StatusOptions) (*postprocesspolicy.StatusSummary, error) { |
| 224 | + return p.store.Status(ctx, opts) |
| 225 | +} |
| 226 | + |
| 227 | +// Reset clears the state of a specific post-processing tool. |
| 228 | +// @intent 실패 누적 상태를 초기화해 특정 후처리 도구를 다시 정상 정책으로 돌린다. |
| 229 | +// @sideEffect 해당 도구의 정책 상태를 재설정한다. |
| 230 | +func (p *MCPPostprocessPolicy) Reset(ctx context.Context, tool string) error { |
| 231 | + return p.store.Reset(ctx, tool) |
| 232 | +} |
0 commit comments