feat: conduit pipelines repair + MCP repair with fix-synthesis (§3/§2)

cmd/conduit/internal/mcp/catalog.go

@@ -87,6 +87,24 @@ var catalog = []ToolInfo{
 		Description: "Scaffolds a new Go (WASM) processor plugin repository. Same engine as `conduit processor new`.",
 		Mutates:     true,
 	},
+	{
+		Name: ToolRepair,
+		Description: "Scans a pipeline configuration for findings that carry a structured, machine-appliable " +
+			"fix (a deprecated/renamed field, an unambiguous invalid status value, a negative processor " +
+			"workers count, an over-long description), classifies each fix's safety (safe / restart / " +
+			"data_path), and returns a plan hash. Mutates nothing. Same engine as `conduit pipelines repair`.",
+		Mutates: false,
+	},
+	{
+		Name: ToolRepairApply,
+		Description: "Applies the plan computed by repair — safe fixes only, unless select names others — " +
+			"only if hash still matches the freshly recomputed plan (a stale hash is refused, nothing " +
+			"mutated). A data-path-adjacent fix (connector settings, a connector's own plugin/type, DLQ " +
+			"config, any id field) is NEVER applied by this tool, even if selected explicitly — it comes " +
+			"back in the result as a skipped fix; only the CLI's --escalate flag (human-only) can apply " +
+			"one. Same engine as `conduit pipelines repair --apply`.",
+		Mutates: true,
+	},
 	{
 		Name: ToolStart,
 		Description: "Starts a pipeline registered in a running Conduit (transitions to Running). " +

cmd/conduit/internal/mcp/server.go

@@ -38,6 +38,8 @@ const (
 	ToolStop              = "stop"
 	ToolScaffoldConnector = "scaffold_connector"
 	ToolScaffoldProcessor = "scaffold_processor"
+	ToolRepair            = "repair"
+	ToolRepairApply       = "repair_apply"
 )
 
 // Config configures NewServer.
@@ -93,18 +95,23 @@ type server struct {
 
 // NewServer builds the `conduit mcp` tool catalog over cfg.
 //
-// Read tools — validate, lint, dry_run, doctor, deploy, inspect — are always
-// registered; none of them mutate anything (deploy computes a Diff/hash but
-// never executes it; see deploy.go).
+// Read tools — validate, lint, dry_run, doctor, deploy, inspect, repair —
+// are always registered; none of them mutate anything (deploy computes a
+// Diff/hash but never executes it, see deploy.go; repair computes a plan
+// over CONTENT and returns proposed fixes, never touching a store or a
+// running pipeline, see cmd/conduit/internal/repair's doc).
 //
-// Write tools — apply, start, stop, scaffold_connector, scaffold_processor —
-// are registered only when cfg.AllowMutations is set (design doc §3,
-// AC-2/AC-3; start/stop added by 20260712-cli-pipeline-lifecycle-verbs.md
-// §4).
+// Write tools — apply, start, stop, scaffold_connector, scaffold_processor,
+// repair_apply — are registered only when cfg.AllowMutations is set (design
+// doc §3, AC-2/AC-3; start/stop added by
+// 20260712-cli-pipeline-lifecycle-verbs.md §4; repair_apply added by
+// 20260712-repair-command.md §5.2 — even when registered, it never applies
+// a data-path-adjacent fix; see tools_repair.go's doc).
 //
 // Every tool is a thin handler over the exact engine the matching CLI verb
 // calls (cmd/conduit/internal/validate, pkg/conduit/check via doctorcheck,
-// cmd/conduit/internal/deploy, pkg/scaffold, the API client) — see doc.go.
+// cmd/conduit/internal/deploy, cmd/conduit/internal/repair, pkg/scaffold,
+// the API client) — see doc.go.
 func NewServer(cfg Config) *sdkmcp.Server {
 	if cfg.newDeployService == nil {
 		cfg.newDeployService = deploy.NewService
@@ -120,11 +127,15 @@ func NewServer(cfg Config) *sdkmcp.Server {
 		Instructions: "Conduit pipeline tools. validate/lint/dry_run/deploy check and preview a pipeline " +
 			"configuration offline (content-in: pass the pipeline YAML as `config`, never a server file path). " +
 			"doctor checks whether the local machine/configuration is ready for `conduit run`. inspect reads a " +
-			"running pipeline's live status (requires the server to have been started with --api-address). If " +
-			"present, apply/start/stop/scaffold_connector/scaffold_processor mutate the local pipeline store, a " +
-			"running pipeline, or the filesystem — apply requires the hash from a prior deploy call and is " +
-			"refused on a stale hash or a running pipeline; start/stop require --api-address like inspect and " +
-			"are refused on an invalid transition (already running / not running); these tools only appear when " +
+			"running pipeline's live status (requires the server to have been started with --api-address). " +
+			"repair scans a configuration for machine-appliable fixes and returns them classified " +
+			"(safe/restart/data_path) with a plan hash; repair_apply, when present, applies the safe ones " +
+			"(or an explicit `select`) but NEVER a data_path fix — those require the human-only CLI --escalate " +
+			"path. If present, apply/start/stop/scaffold_connector/scaffold_processor/repair_apply mutate the " +
+			"local pipeline store, a running pipeline, or the filesystem — apply requires the hash from a prior " +
+			"deploy call and is refused on a stale hash or a running pipeline; start/stop require --api-address " +
+			"like inspect and are refused on an invalid transition (already running / not running); these tools " +
+			"(plus repair_apply) only appear when " +
 			"the operator started conduit mcp with --allow-mutations.",
 	})
 
@@ -134,13 +145,15 @@ func NewServer(cfg Config) *sdkmcp.Server {
 	sdkmcp.AddTool(srv, tool(ToolDoctor), s.doctor)
 	sdkmcp.AddTool(srv, tool(ToolDeploy), s.deploy)
 	sdkmcp.AddTool(srv, tool(ToolInspect), s.inspect)
+	sdkmcp.AddTool(srv, tool(ToolRepair), s.repair)
 
 	if cfg.AllowMutations {
 		sdkmcp.AddTool(srv, tool(ToolApply), s.apply)
 		sdkmcp.AddTool(srv, tool(ToolStart), s.start)
 		sdkmcp.AddTool(srv, tool(ToolStop), s.stop)
 		sdkmcp.AddTool(srv, tool(ToolScaffoldConnector), s.scaffoldConnector)
 		sdkmcp.AddTool(srv, tool(ToolScaffoldProcessor), s.scaffoldProcessor)
+		sdkmcp.AddTool(srv, tool(ToolRepairApply), s.repairApply)
 	}
 
 	return srv

cmd/conduit/internal/mcp/server_test.go

@@ -26,9 +26,9 @@ import (
 	sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp"
 )
 
-var readToolNames = []string{ToolValidate, ToolLint, ToolDryRun, ToolDoctor, ToolDeploy, ToolInspect}
+var readToolNames = []string{ToolValidate, ToolLint, ToolDryRun, ToolDoctor, ToolDeploy, ToolInspect, ToolRepair}
 
-var writeToolNames = []string{ToolApply, ToolStart, ToolStop, ToolScaffoldConnector, ToolScaffoldProcessor}
+var writeToolNames = []string{ToolApply, ToolStart, ToolStop, ToolScaffoldConnector, ToolScaffoldProcessor, ToolRepairApply}
 
 // TestNewServer_ReadToolsAlwaysRegistered is AC-1's catalog half: every read
 // tool is present regardless of AllowMutations.

cmd/conduit/internal/mcp/tools_repair.go

@@ -0,0 +1,92 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mcp
+
+import (
+	"context"
+
+	"github.com/conduitio/conduit/cmd/conduit/internal/repair"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr"
+	sdkmcp "github.com/modelcontextprotocol/go-sdk/mcp"
+)
+
+// RepairArgs is the repair (read) tool's input: the pipeline configuration
+// YAML CONTENT to scan for machine-appliable fixes.
+type RepairArgs struct {
+	Config string `json:"config" jsonschema:"the pipeline configuration YAML content to scan for machine-appliable fixes (a single pipeline document)"`
+}
+
+// repair implements the repair (read) tool: run the same
+// cmd/conduit/internal/repair.CollectContent engine `conduit pipelines
+// repair` uses over the content, and return the proposed fixes, their
+// safety classification, and a plan hash. Mutates nothing — parity with
+// the always-on validate/lint/dry_run/deploy tools. Same engine as
+// `conduit pipelines repair`.
+func (s *server) repair(ctx context.Context, _ *sdkmcp.CallToolRequest, in RepairArgs) (*sdkmcp.CallToolResult, Result[repair.Plan], error) {
+	plan, err := repair.CollectContent(ctx, in.Config)
+	if err != nil {
+		return toolErr[repair.Plan](err)
+	}
+	return toolOK(true, plan, repair.SummarizePlan(plan))
+}
+
+// RepairApplyArgs is the repair_apply tool's input: the same content the
+// repair tool was called with, the plan Hash from repair's result, and an
+// optional Select of specific configPaths to apply.
+//
+// There is deliberately NO Escalate field here (design doc §5.2, AC-15):
+// repair.ApplyInput.Escalate — the human-only Tier-1 override that permits
+// applying a FixClassDataPath fix — is reachable only from the CLI's
+// --escalate flag (cmd/conduit/root/pipelines/repair.go). An agent that
+// explicitly Selects a data-path-adjacent fix's configPath still gets it
+// back as FixOutcomeSkipped with repair.CodeDataPathFixRefused in the
+// result's Fixes report — repair_apply NEVER applies one, and there is no
+// schema field an agent could set to force it (mirrors AllowMutations
+// being process-set only, server.go's Config doc).
+type RepairApplyArgs struct {
+	Config string `json:"config" jsonschema:"the pipeline configuration YAML content to repair (must match what the repair tool scanned)"`
+	// Hash is schema-optional (like ApplyArgs.Hash) so the handler's own
+	// check below gives an actionable conduiterr-mapped error rather than
+	// the SDK's generic schema-validation error.
+	Hash   string   `json:"hash,omitempty" jsonschema:"the plan hash from the repair tool's result; repair_apply refuses to run unless this matches the freshly recomputed plan hash exactly"`
+	Select []string `json:"select,omitempty" jsonschema:"apply only the fix(es) at these configPaths; omit to apply every safe fix (the default)"`
+}
+
+// repairApply implements the repair_apply (write) tool: recompute the plan
+// from Config and apply it only if Hash matches the freshly recomputed
+// plan's hash exactly (design doc §5.2, mirroring the apply tool's own
+// stance — no "skip the hash" escape hatch for MCP). Applies FixClassSafe
+// fixes by default (or the Select-ed subset); a FixClassDataPath fix is
+// never applied — see RepairApplyArgs' doc. Only registered when the
+// server was started with --allow-mutations. Same engine as `conduit
+// pipelines repair --apply`.
+func (s *server) repairApply(ctx context.Context, _ *sdkmcp.CallToolRequest, in RepairApplyArgs) (*sdkmcp.CallToolResult, Result[repair.Result], error) {
+	if in.Hash == "" {
+		ce := conduiterr.New(conduiterr.CodeInvalidArgument, "hash is required (from the repair tool's result)")
+		ce.Suggestion = "call the repair tool first, review its proposed fixes, then pass its hash to repair_apply"
+		return toolErr[repair.Result](ce)
+	}
+
+	res, err := repair.Apply(ctx, repair.ApplyInput{
+		Content: in.Config,
+		Hash:    in.Hash,
+		Select:  in.Select,
+		// Escalate is always false here — see RepairApplyArgs' doc.
+	})
+	if err != nil {
+		return toolErr[repair.Result](err)
+	}
+	return toolOK(true, res, repair.SummarizeResult(res))
+}