feat(api): idempotent test creation and an sla on flat requests

cli/cmd/mcp_caveats.go

@@ -76,7 +76,7 @@ var mcpCaveatsCore = []mcpCaveat{
 		Short: "A LOAD run is one producer and one consumer whatever numProducers says; only STRESS runs parallel producers.",
 		Text: "A LOAD run is one producer and one consumer, whatever numProducers and numConsumers say, " +
 			"so it cannot show how the cluster behaves under parallel clients. STRESS starts one producer per numProducers.",
-		Refs: []string{mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1656-1665", "case LOAD ->", "case STRESS ->")},
+		Refs: []string{mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1677-1686", "case LOAD ->", "case STRESS ->")},
 	},
 	{
 		ID:    mcpCaveatReaperDeadline,
@@ -99,11 +99,11 @@ var mcpCaveatsCore = []mcpCaveat{
 			"kates/src/main/resources/application.properties:333-347",
 			mcpAnchoredRef(mcpJava+"engine/TestTimeoutReaper.java:49-51,67-76,161-164,192-205",
 				"TestResult.TaskStatus.STOPPING", "kates.engine.reaper-grace-ms", "\"-submission\"", "\"Timeout: still \""),
-			mcpJava + "engine/TestOrchestrator.java:132-154,186-189,354,1125-1184,1250-1259,1314-1346",
-			mcpJava + "persistence/EntityMapper.java:158-167",
+			mcpJava + "engine/TestOrchestrator.java:132-163,197-200,366,1142-1205,1271-1280,1335-1367",
+			mcpJava + "persistence/EntityMapper.java:161-170",
 			"kates/src/main/resources/db/migration/V24__run_planned_duration.sql:1-13",
-			mcpJava + "domain/TestRun.java:478-481",
-			mcpJava + "api/HealthResource.java:51-72",
+			mcpJava + "domain/TestRun.java:520-523",
+			mcpJava + "api/HealthResource.java:53-75",
 		},
 	},
 	{
@@ -123,13 +123,13 @@ var mcpCaveatsCore = []mcpCaveat{
 			"integrity-cg-integrity, without transactions.",
 		Refs: []string{
 			mcpJava + "domain/TestSpec.java:14-32,111-113",
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:165,180-182,225",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:174,189-191,236",
 				"applyTypeDefaults(type, request.getSpec())", ".withRequestedSpec(", "buildTasks(type, spec, run.getId())"),
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1045-1091", "TestSpec applyTypeDefaults(", "return merged;"),
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1635-1639,1730-1734",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1062-1108", "TestSpec applyTypeDefaults(", "return merged;"),
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1656-1660,1751-1755",
 				"Only when asked", `"-integrity-0-group")`),
 			mcpAnchoredRef(mcpJava+"engine/NativeKafkaBackend.java:792", `task.getConsumerGroup() + "-integrity"`),
-			mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:56,84-85,143-145",
+			mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:56,85-86,146-148",
 				"entity.setRequestedSpecJson(toJson(run.getRequestedSpec()))", "fromJson(entity.getRequestedSpecJson()",
 				"if (run.getRequestedSpec() != null)"),
 			"kates/src/main/resources/db/migration/V23__requested_spec_fields.sql:1-6",
@@ -174,7 +174,7 @@ var mcpCaveatsCore = []mcpCaveat{
 		Text: "A TUNE_* run executes one produce task with the spec's single configuration: it measures one setting " +
 			"and compares none.",
 		Refs: []string{
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1741-1742", "case TUNE_REPLICATION,", "-tune-0"),
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1762-1763", "case TUNE_REPLICATION,", "-tune-0"),
 			mcpJava + "trogdor/SpecFactory.java:38-39",
 		},
 	},
@@ -186,7 +186,7 @@ var mcpCaveatsCore = []mcpCaveat{
 		Refs: []string{
 			mcpJava + "trend/TrendResource.java:31-60",
 			mcpJava + "trend/TrendService.java:53-54,185-189",
-			mcpAnchoredRef(mcpJava+"service/TestRunRepository.java:341-361",
+			mcpAnchoredRef(mcpJava+"service/TestRunRepository.java:351-371",
 				"List<TestRun> findWithResults(", "WHERE r.testType = :type AND r.status = :status",
 				".map(EntityMapper::toDomain)"),
 		},

cli/cmd/mcp_guard.go

@@ -125,7 +125,7 @@ func mcpInvalidArgument(message, value string) error {
 
 // mcpID is a test-run or disruption id as tool input. The backend mints both
 // as the first 8 characters of a random UUID, so 8 lowercase hex characters
-// (domain/TestRun.java:46,64; disruption/DisruptionLauncher.java:77). The
+// (domain/TestRun.java:48,67; disruption/DisruptionLauncher.java:77). The
 // schema carries the pattern, so the SDK refuses a bad id before any Kates
 // code runs; mcpValidateID checks again for ids that arrive another way.
 type mcpID string

cli/cmd/mcp_prompts_cluster.go

@@ -95,7 +95,7 @@ func mcpDidKatesCauseThis(_ context.Context, req *mcp.GetPromptRequest) (*mcp.Ge
 
 // mcpDidKatesCauseThisText is the prompt itself. Every fact it states about
 // Kates was checked against the code: runs and disruptions record no owner
-// (domain/TestRun.java:16-42, disruption/DisruptionReportEntity.java:15-34);
+// (domain/TestRun.java:16-44, disruption/DisruptionReportEntity.java:15-34);
 // audit rows name the principal that made each change (V26), and every REST
 // and gRPC change and every scheduled firing writes one
 // (audit/AuditResponseFilter.java, security/GrpcApiKeyInterceptor.java,

cli/cmd/mcp_tools_runs_assess.go

@@ -361,7 +361,7 @@ func mcpRunHasRequested(r *client.MCPRun) bool {
 
 // mcpAssessBaseline finds the baseline run: among the runs the band's scan
 // read, or else with one read of its own. The baseline may be any run, of
-// any type or status (TestResource.java:590-618 checks only that it exists),
+// any type or status (TestResource.java:659-687 checks only that it exists),
 // so reading it polls it if it is still active, as get_run does.
 func mcpAssessBaseline(ctx context.Context, call *mcpCall, baselineID string, band mcpBandScan) *mcpRunIdentity {
 	if b, ok := band.byID[baselineID]; ok {

cli/cmd/mcp_tools_runs_list.go

@@ -26,7 +26,7 @@ import (
 // Every tool goes through addReadTool and reads only with GET. get_run and
 // assess_run read the run with GET /api/tests/{id}, which makes the backend
 // poll a run that is still active and save what it finds
-// (TestResource.java:458-464, TestOrchestrator.refreshStatus); both
+// (TestResource.java:527-533, TestOrchestrator.refreshStatus); both
 // descriptions say so. Third-party text (task errors, scenario and phase
 // names, labels, plan names, audit details, backend messages) is fenced;
 // ids, types and statuses are cleaned so an agent can pass them back.
@@ -574,15 +574,15 @@ func mcpRunDecodeSpec(raw json.RawMessage, w *mcpRunSpecWire) (bool, error) {
 }
 
 // mcpRunSpecFrom reads the stored spec. A run without a topic used one named
-// after its type (TestOrchestrator.java:1614,1941-1942). The topic, acks and
+// after its type (TestOrchestrator.java:1635,1962-1963). The topic, acks and
 // compressionType are shown as identifiers only when they hold values Kafka
 // accepts (mcpKafkaTopicRE, mcpAcksRE, mcpCompressionRE, the values
 // TestSpec's bean validation accepts). A stored run can hold others: a
 // schedule saved by a PUT, which runs no bean validation
 // (ScheduleResource.java:224-226), fires its spec as sent, a Kates API
-// without the check of a scenario's specs (TestOrchestrator.java:1126-1130)
-// stored a scenario's base spec as sent (TestOrchestrator.java:347-355,
-// 373-377), and one without the check of a gRPC request's fields
+// without the check of a scenario's specs (TestOrchestrator.java:1143-1147)
+// stored a scenario's base spec as sent (TestOrchestrator.java:358-367,
+// 385-389), and one without the check of a gRPC request's fields
 // (GrpcTestService.java:116-121) stored a compressionType sent over gRPC
 // unchecked. So another value is third-party text and goes, fenced, into
 // invalid. The seven fields a backend without
@@ -706,7 +706,7 @@ func mcpRunRequestedFrom(call *mcpCall, run *client.MCPRun) (*mcpRunRequestedSpe
 func mcpRunTaskFrom(call *mcpCall, t client.MCPRunTask) mcpRunTaskOut {
 	return mcpRunTaskOut{
 		// A scenario run's task ids are the run id and the phase name
-		// (TestOrchestrator.java:1758), which nothing validates.
+		// (TestOrchestrator.java:1779), which nothing validates.
 		TaskID:              call.FenceN(t.TaskID, 128),
 		Phase:               call.FenceN(t.PhaseName, 64),
 		Status:              mcpSanitizeLine(t.Status, 16),
@@ -1031,7 +1031,7 @@ var mcpCaveatsRunsList = []mcpCaveat{
 			mcpAnchoredRef(mcpJava+"report/ReportGenerator.java:220-249,283-294",
 				"report.setSummary(MetricUtils.computeSummary(results));",
 				"summaries.put(phase, MetricUtils.computeSummary(rows))"),
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:415-424,2089-2105",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:427-436,2110-2126",
 				"TestResult.TaskStatus.PENDING : TestResult.TaskStatus.RUNNING",
 				"private static TestRun withUnfinishedTasksFailed("),
 		},
@@ -1051,7 +1051,7 @@ var mcpCaveatsRunsList = []mcpCaveat{
 				"private void runIntegrity(", "state.integrityResult = state.verifier.verify("),
 			mcpAnchoredRef(mcpJava+"engine/DataIntegrityVerifier.java:51,224-227",
 				"static final int MAX_LOST_RANGES = 1000;", "if (chaosStartNanos <= 0) {"),
-			mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:34-43,216-223,243",
+			mcpAnchoredRef(mcpJava+"persistence/EntityMapper.java:34-43,219-226,246",
 				"entity.setIntegrityJson(integrityJson);",
 				".withIntegrity(fromJson(INTEGRITY_JSON, entity.getIntegrityJson(), IntegrityResult.class));",
 				"DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false"),
@@ -1070,9 +1070,9 @@ var mcpCaveatsRunsList = []mcpCaveat{
 			"except for a run cancelled before its tasks existed; a cancel through the REST API also leaves a CANCEL " +
 			"audit row.",
 		Refs: []string{
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:2040-2118",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:2061-2139",
 				`"Cancelled by user"`, `EventKind.FAILED, "cancelled"`, "return run.withResults(updatedResults);"),
-			mcpAnchoredRef(mcpJava+"api/TestResource.java:497-542",
+			mcpAnchoredRef(mcpJava+"api/TestResource.java:566-611",
 				`@Path("/{id}/cancel")`, `auditService.record("CANCEL"`, `"Test cancelled; it is stored as FAILED"`),
 			mcpJava + "domain/TestResult.java:25-31",
 		},
@@ -1100,14 +1100,14 @@ var mcpCaveatsRunsList = []mcpCaveat{
 			"say which the API is, but a scenario run's tasks do: on an older API they all start within moments " +
 			"of one another.",
 		Refs: []string{
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:341-343,391-402,404-420",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:352-354,403-414,416-432",
 				"applyTypeDefaults(type, scenario.getBaseSpec())", ".withRequestedSpec(",
 				"long phaseStartMs = System.currentTimeMillis();", "scenario.resolveSpecForPhase(phase)",
 				"phaseStartMs = saturatedSum(phaseStartMs", "backend.submit(task)", "TestResult.TaskStatus.PENDING"),
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1452-1480,1552-1591,1601-1613",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1473-1501,1573-1612,1622-1634",
 				"MAX_RAMP_STEPS = 100", "scenarioInapplicableFields(TestScenario scenario", "is a SPIKE phase",
 				"has none (", "check no record CRCs"),
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1738-1807", "List<BenchmarkTask> buildPhaseTask(",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1759-1828", "List<BenchmarkTask> buildPhaseTask(",
 				"int baseTarget = Math.max(1, spec.getThroughput() / steps);", "saturatedSum(startAtMs, s * stepMs)",
 				`taskId + "-spike"`),
 			mcpAnchoredRef(mcpJava+"domain/TestScenario.java:108-183",

cli/cmd/mcp_tools_runs_test.go

@@ -813,9 +813,9 @@ func mcpOutsideFences(h *mcpHarness, raw []byte) string {
 }
 
 // A scenario run's text reaches the store unvalidated: its phase names
-// become task ids (TestOrchestrator.java:1758), and an older Kates API
+// become task ids (TestOrchestrator.java:1779), and an older Kates API
 // stored its base spec's topic, acks and compression as sent
-// (TestOrchestrator.java:347-355,373-377). None of it may reach the model outside a
+// (TestOrchestrator.java:358-367,385-389). None of it may reach the model outside a
 // fence.
 func TestMCPGetRunUnvalidatedScenarioText(t *testing.T) {
 	fb := newMCPFakeBackend(t, "cluster-a")

cli/cmd/mcp_tools_scenario.go

@@ -1240,9 +1240,9 @@ var mcpCaveatsScenario = []mcpCaveat{
 			"and kates test apply sends a scenario with whatever key it holds.",
 		Refs: []string{
 			"plans/mcp-server.md:387-405",
-			mcpJava + "api/TestResource.java:108-135",
+			mcpJava + "api/TestResource.java:124-192",
 			mcpJava + "domain/TestSpec.java:34-113",
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1372-1447",
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1393-1468",
 				"Map<String, String> inapplicableFields(", "the trogdor backend cannot run a transactional producer"),
 			mcpAnchoredRef("cli/cmd/apply.go:238-307",
 				"for i, scenario := range sf.Scenarios {", "apiClient.CreateTest(a.ctx, scenarioToRequest(scenario))",
@@ -1259,7 +1259,7 @@ var mcpCaveatsScenario = []mcpCaveat{
 			"listed as defaulted may differ on the cluster that runs the scenario.",
 		Refs: []string{
 			mcpJava + "config/TestTypeDefaults.java:22-64,323-464",
-			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1045-1091", "TestSpec applyTypeDefaults(", "return merged;"),
+			mcpAnchoredRef(mcpJava+"engine/TestOrchestrator.java:1062-1108", "TestSpec applyTypeDefaults(", "return merged;"),
 			mcpAnchoredRef("kates/src/main/resources/application.properties:73-114",
 				"# Per-type overrides: STRESS", "kates.tests.roundtrip.throughput=10000"),
 			mcpAnchoredRef("charts/kates/values.yaml:691-759", "defaults:", "tests:", `throughput: "10000"`),

cli/schemas/schema_test.go

@@ -411,6 +411,11 @@ func TestSchemasRejectInvalidEnumsTypesAndBounds(t *testing.T) {
 			schema: "test-request.v1.schema.json",
 			doc:    `{"scenario": {"name": "s", "phases": [{"name": "p1", "type": "RAMP", "durationMs": 1000, "rampSteps": 1}]}}`,
 		},
+		{
+			name:   "test-request sla beside a scenario",
+			schema: "test-request.v1.schema.json",
+			doc:    `{"sla": {"maxP99LatencyMs": 1}, "scenario": {"name": "s", "phases": [{"name": "p1", "type": "STEADY", "durationMs": 1000}]}}`,
+		},
 		{
 			name:   "flow invalid action enum",
 			schema: "flow.v1.schema.json",

cli/schemas/test-request.v1.schema.json

@@ -9,6 +9,7 @@
     { "required": ["type"] },
     { "required": ["scenario"] }
   ],
+  "not": { "required": ["sla", "scenario"] },
   "properties": {
     "type": {
       "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).",
@@ -22,6 +23,10 @@
       "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).",
       "$ref": "#/$defs/wireTestSpec"
     },
+    "sla": {
+      "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.",
+      "$ref": "#/$defs/slaDefinition"
+    },
     "scenario": {
       "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.",
       "$ref": "#/$defs/multiPhaseScenario"

editors/visualstudio-kates/Schemas/test-request.v1.schema.json

@@ -9,6 +9,7 @@
     { "required": ["type"] },
     { "required": ["scenario"] }
   ],
+  "not": { "required": ["sla", "scenario"] },
   "properties": {
     "type": {
       "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).",
@@ -22,6 +23,10 @@
       "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).",
       "$ref": "#/$defs/wireTestSpec"
     },
+    "sla": {
+      "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.",
+      "$ref": "#/$defs/slaDefinition"
+    },
     "scenario": {
       "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.",
       "$ref": "#/$defs/multiPhaseScenario"

editors/vscode-kates/schemas/test-request.v1.schema.json

@@ -9,6 +9,7 @@
     { "required": ["type"] },
     { "required": ["scenario"] }
   ],
+  "not": { "required": ["sla", "scenario"] },
   "properties": {
     "type": {
       "description": "Test type for a single-phase run (or fallback if scenario.type is omitted).",
@@ -22,6 +23,10 @@
       "description": "Wire-format TestSpec parameters (uses numRecords, numProducers, recordSize, durationMs, throughput).",
       "$ref": "#/$defs/wireTestSpec"
     },
+    "sla": {
+      "description": "Pass/fail criteria for a run with no scenario; a scenario carries its own under scenario.sla.",
+      "$ref": "#/$defs/slaDefinition"
+    },
     "scenario": {
       "description": "Multi-phase scenario envelope with baseSpec, ordered execution phases, SLA definition, and metadata labels.",
       "$ref": "#/$defs/multiPhaseScenario"

kates/src/main/java/com/bmscomp/kates/api/HealthResource.java

@@ -1,6 +1,7 @@
 package com.bmscomp.kates.api;
 
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import jakarta.annotation.security.PermitAll;
 import jakarta.inject.Inject;
@@ -47,12 +48,14 @@ public HealthResource(
     @GET
     @Operation(
             summary = "Application health",
-            description = "Returns engine status, Kafka connectivity, and test type configurations")
+            description = "Returns engine status, Kafka connectivity, test type configurations, and the"
+                    + " capabilities a client can check before relying on them")
     public Map<String, Object> health() {
         boolean kafkaReachable = clusterHealthService.isReachable();
 
         Map<String, Object> response = new LinkedHashMap<>();
         response.put("status", kafkaReachable ? "UP" : "DEGRADED");
+        response.put("capabilities", List.of("clientKey"));
 
         response.put(
                 "engine",

kates/src/main/java/com/bmscomp/kates/api/KafkaClientResource.java

@@ -217,6 +217,10 @@ public Response createTopic(CreateTopicRequest request) {
             topicService.createTopic(request.name(), partitions, rf, request.configs());
             var detail = topicService.describeCreatedTopic(request.name());
             return Response.status(201).entity(detail).build();
+        } catch (IllegalArgumentException e) {
+            return Response.status(400)
+                    .entity(ApiError.of(400, "Bad Request", e.getMessage()))
+                    .build();
         } catch (Exception e) {
             return Response.serverError()
                     .entity(ApiError.of(500, "Kafka Error", e.getMessage()))