From 4a36f6504dbb2bebf45badec7d0b197515d0bf8b Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:09:06 -0400 Subject: [PATCH 1/6] test(fixtures): add golden scenario fixtures and schema validation tests --- test/fixtures/decisions/cart-failure.json | 246 ++++++++++++++++ test/fixtures/detections/cart-failure.json | 55 ++++ test/fixtures/fixtures_test.go | 276 ++++++++++++++++++ test/fixtures/proposals/cart-failure.json | 230 +++++++++++++++ .../fixtures/telemetry/loki-cart-failure.json | 20 ++ .../fixtures/telemetry/prom-cart-failure.json | 18 ++ 6 files changed, 845 insertions(+) create mode 100644 test/fixtures/decisions/cart-failure.json create mode 100644 test/fixtures/detections/cart-failure.json create mode 100644 test/fixtures/fixtures_test.go create mode 100644 test/fixtures/proposals/cart-failure.json create mode 100644 test/fixtures/telemetry/loki-cart-failure.json create mode 100644 test/fixtures/telemetry/prom-cart-failure.json diff --git a/test/fixtures/decisions/cart-failure.json b/test/fixtures/decisions/cart-failure.json new file mode 100644 index 0000000..d42f46c --- /dev/null +++ b/test/fixtures/decisions/cart-failure.json @@ -0,0 +1,246 @@ +{ + "decision": { + "id": "dec:slo_burn:cart:1785671212", + "proposalRef": "cart-burn-accel", + "signalRef": "slo_burn:cart", + "sloRef": "cart-availability", + "candidateRef": "p1", + "verdict": "approved", + "requestedBand": "act_reversible", + "grantedBand": "act_reversible", + "riskBand": "act_reversible", + "floorApplied": 0.75, + "policyVersion": "v1", + "evaluatedAt": "2026-08-02T11:46:52Z" + }, + "set": { + "name": "cart-burn-accel", + "runID": "slo_burn:cart/1786969860387330045", + "signalRef": "slo_burn:cart", + "sloRef": "cart-availability", + "saoSnapshot": { + "version": 1, + "assembledAt": "2026-08-02T11:46:10Z", + "signal": { + "confidence": 1, + "severity": { + "degradationPct": 0.5, + "trajectory": "accelerating" + }, + "blastRadius": { + "affectedPct": 0.12, + "velocity": "accelerating", + "downstreamConsumers": 2 + }, + "originService": "cart" + }, + "topology": { + "upstream": [ + { + "name": "flagd", + "state": "healthy", + "trafficShare": 0.03 + }, + { + "name": "redis", + "state": "healthy", + "trafficShare": 0.89 + }, + { + "name": "unknown", + "state": "degraded", + "trafficShare": 0.08 + } + ], + "downstream": [ + { + "name": "checkout", + "state": "degraded", + "trafficShare": 0.2 + }, + { + "name": "frontend", + "state": "healthy", + "trafficShare": 0.8 + } + ] + }, + "change": { + "events": [ + { + "id": "config/otel-demo/flagd-config/4283", + "type": "config", + "target": "flagd", + "age": 289392211635, + "historicalStaleness": 289392211635 + }, + { + "id": "deploy/otel-demo/checkout/1", + "type": "deploy", + "target": "checkout", + "age": 580392211635, + "historicalStaleness": 580392211635 + }, + { + "id": "deploy/otel-demo/cart/1", + "type": "deploy", + "target": "cart", + "age": 593392211635, + "historicalStaleness": 593392211635 + } + ] + } + }, + "failureClass": "service_failure", + "causalScores": [ + { + "eventID": "config/otel-demo/flagd-config/4283", + "inTopology": true, + "temporal": 0.89, + "historical": 0.9, + "liveCorroborated": true, + "likelihood": 0.85, + "rationale": [ + "temporal: 4m49s old -> 0.89", + "topological: in-path=true -> 0.20", + "historical: case-base prior -> 0.90" + ] + }, + { + "eventID": "deploy/otel-demo/checkout/1", + "inTopology": true, + "temporal": 0.8, + "topological": 0.2, + "historical": 0.9, + "liveCorroborated": true, + "likelihood": 0.63, + "rationale": [ + "temporal: 9m40s old -> 0.80", + "topological: in-path=true -> 0.20", + "historical: case-base prior -> 0.90" + ] + }, + { + "eventID": "deploy/otel-demo/cart/1", + "temporal": 0.8, + "historical": 0.9, + "likelihood": 0.5, + "rationale": [ + "temporal: 9m53s old -> 0.80", + "topological: in-path=false -> 0.00", + "historical: case-base prior -> 0.90", + "defence 1: uncorroborated -> capped at 0.50", + "target \"cart\" is not in this signal's topology -> scored for audit, excluded from confidence" + ] + } + ], + "hypotheses": [ + { + "name": "cartFailure flag injected via recent flagd config change", + "weight": 0.85 + }, + { + "name": "Cart pod crash loop or transient restart issue", + "weight": 0.1 + }, + { + "name": "Redis connectivity degradation affecting cart state operations", + "weight": 0.05 + } + ], + "evidence": [ + { + "tool": "metrics", + "query": "cart_error_ratio", + "key": "cart_error_ratio", + "summary": "cart_error_ratio = 0.50", + "ref": "metrics://cart_error_ratio", + "live": true, + "subject": "cart" + }, + { + "tool": "metrics", + "query": "severity_cart_availability", + "key": "severity_cart_availability", + "summary": "severity_cart_availability = 0.50", + "ref": "metrics://severity_cart_availability", + "live": true, + "subject": "cart" + }, + { + "tool": "kube", + "query": "{\"resource\":\"pods\",\"namespace\":\"otel-demo\",\"selector\":{\"app.kubernetes.io/name\":\"cart\"}}", + "key": "kube-1", + "summary": "cart-79bf56b7f9-r5ngh (Running)", + "ref": "kube://otel-demo/pods", + "live": true, + "subject": "cart" + }, + { + "tool": "loki", + "query": "{namespace=\"otel-demo\", service_name=\"cart\"}", + "key": "loki-1", + "summary": "50 log line(s); last: rpc error: code = Unavailable desc = connection error", + "ref": "loki://otel-demo/service_name=cart", + "live": true, + "subject": "cart" + } + ], + "serviceTier": "tier-1", + "gate": { + "budgetOK": true, + "dedupeOK": true, + "evidenceOK": true, + "passed": true + }, + "proposals": [ + { + "id": "p1", + "contractRef": "disable-cart-failure", + "confidence": 0.85, + "computedConfidence": 1, + "confidenceCeilingBound": true, + "terms": { + "signalConfidence": 1, + "corroborated": 3, + "grounding": 1, + "alignmentOK": true, + "likelihoodOK": true + }, + "predictedImpact": { + "severityReductionPct": 0.9, + "blastRadiusDelta": 0, + "sloEffects": { + "time_to_effect": "30s" + } + }, + "blastTier": "low", + "reversalPath": { + "method": "enable-cart-failure", + "watching": "cart_error_ratio", + "trigger": "cart_error_ratio < 0.02", + "automatic": true + }, + "governanceLevel": { + "band": "act_reversible" + }, + "rank": 1, + "citations": [ + "cart_error_ratio", + "severity_cart_availability", + "kube-1", + "loki-1" + ] + } + ], + "recommended": "p1", + "rankingRationale": { + "dominantAxis": "time_to_effect", + "velocityWeight": "accelerating" + }, + "status": { + "phase": "proposed", + "observedAt": "2026-08-02T11:46:30Z" + } + } +} diff --git a/test/fixtures/detections/cart-failure.json b/test/fixtures/detections/cart-failure.json new file mode 100644 index 0000000..f1aa830 --- /dev/null +++ b/test/fixtures/detections/cart-failure.json @@ -0,0 +1,55 @@ +{ + "name": "cart-burn-accel", + "fingerprint": "slo_burn:cart", + "originService": "cart", + "serviceTier": "tier-1", + "detectorType": "sustained_burn", + "divergence": { + "metric": "cart_error_ratio", + "observed": 0.5, + "baseline": 0.0, + "confidence": 1.0, + "trajectory": "accelerating" + }, + "topology": { + "upstream": [ + { + "name": "flagd", + "state": "healthy" + }, + { + "name": "redis", + "state": "healthy" + } + ], + "downstream": [ + { + "name": "checkout", + "state": "degraded" + }, + { + "name": "frontend", + "state": "healthy" + } + ] + }, + "traffic": { + "affectedPct": 0.12, + "baseline": 100.0, + "baselineWindow": "1h" + }, + "impact": { + "severity": { + "degradationPct": 0.5, + "trajectory": "accelerating" + }, + "blastRadius": { + "affectedPct": 0.12, + "velocity": "accelerating", + "downstreamConsumers": 2 + } + }, + "contractRef": "cart-availability:v1", + "sloRef": "cart-availability", + "detectedAt": "2026-08-02T11:46:06.785322491Z" +} diff --git a/test/fixtures/fixtures_test.go b/test/fixtures/fixtures_test.go new file mode 100644 index 0000000..969cadc --- /dev/null +++ b/test/fixtures/fixtures_test.go @@ -0,0 +1,276 @@ +package fixtures_test + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/api/v1/signal" +) + +type promResponse struct { + Status string `json:"status"` + Data struct { + ResultType string `json:"resultType"` + Result []struct { + Metric map[string]string `json:"metric"` + Value []any `json:"value"` + } `json:"result"` + } `json:"data"` +} + +type lokiResponse struct { + Status string `json:"status"` + Data struct { + ResultType string `json:"resultType"` + Result []struct { + Stream map[string]string `json:"stream"` + Values [][]string `json:"values"` + } `json:"result"` + } `json:"data"` +} + +func TestFixtures_SchemaValidity(t *testing.T) { + t.Parallel() + + t.Run("Detection fixtures unmarshal into valid signal.Detection structs", func(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + relPath string + wantFingerprint string + wantService string + wantSLORef string + wantMetric string + }{ + "UnmarshalDetection unmarshals cart-failure JSON into a valid signal.Detection": { + relPath: filepath.Join("detections", "cart-failure.json"), + wantFingerprint: "slo_burn:cart", + wantService: "cart", + wantSLORef: "cart-availability", + wantMetric: "cart_error_ratio", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + data, err := os.ReadFile(tc.relPath) //nolint:gosec // G304: fixed testdata path, not user input + if err != nil { + t.Fatalf("failed to read fixture file %s: %v", tc.relPath, err) + } + + var det signal.Detection + if err := json.Unmarshal(data, &det); err != nil { + t.Fatalf("failed to unmarshal signal.Detection from %s: %v", tc.relPath, err) + } + + if diff := cmp.Diff(tc.wantFingerprint, det.Fingerprint); diff != "" { + t.Errorf("wrong detection fingerprint (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantService, det.OriginService); diff != "" { + t.Errorf("wrong detection origin service (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantSLORef, det.SLORef); diff != "" { + t.Errorf("wrong detection sloRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantMetric, det.Divergence.Metric); diff != "" { + t.Errorf("wrong detection divergence metric (-want +got):\n%s", diff) + } + }) + } + }) + + t.Run("Proposal fixtures unmarshal into valid proposal.Set structs", func(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + relPath string + wantSignalRef string + wantRecommended string + wantClass proposal.FailureClass + wantContractRef string + wantGatePassed bool + }{ + "UnmarshalProposal unmarshals cart-failure JSON into a valid proposal.Set": { + relPath: filepath.Join("proposals", "cart-failure.json"), + wantSignalRef: "slo_burn:cart", + wantRecommended: "p1", + wantClass: proposal.ClassServiceFailure, + wantContractRef: "disable-cart-failure", + wantGatePassed: true, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + data, err := os.ReadFile(tc.relPath) //nolint:gosec // G304: fixed testdata path, not user input + if err != nil { + t.Fatalf("failed to read fixture file %s: %v", tc.relPath, err) + } + + var set proposal.Set + if err := json.Unmarshal(data, &set); err != nil { + t.Fatalf("failed to unmarshal proposal.Set from %s: %v", tc.relPath, err) + } + + if diff := cmp.Diff(tc.wantSignalRef, set.SignalRef); diff != "" { + t.Errorf("wrong proposal SignalRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantRecommended, set.Recommended); diff != "" { + t.Errorf("wrong proposal Recommended candidate (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantClass, set.FailureClass); diff != "" { + t.Errorf("wrong proposal FailureClass (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantContractRef, set.ContractRefFor(set.Recommended)); diff != "" { + t.Errorf("wrong candidate ContractRef (-want +got):\n%s", diff) + } + if set.Gate == nil { + t.Fatal("want non-nil Gate on proposal.Set, got nil") + } + if diff := cmp.Diff(tc.wantGatePassed, set.Gate.Passed); diff != "" { + t.Errorf("wrong gate passed status (-want +got):\n%s", diff) + } + }) + } + }) + + t.Run("Decision fixtures unmarshal into valid decision.Governed envelopes", func(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + relPath string + wantSignalRef string + wantVerdict decision.Verdict + wantBand decision.Band + wantPolicyVer string + }{ + "UnmarshalDecision unmarshals cart-failure JSON into an auditable decision.Governed": { + relPath: filepath.Join("decisions", "cart-failure.json"), + wantSignalRef: "slo_burn:cart", + wantVerdict: decision.VerdictApproved, + wantBand: decision.BandActReversible, + wantPolicyVer: "v1", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + data, err := os.ReadFile(tc.relPath) //nolint:gosec // G304: fixed testdata path, not user input + if err != nil { + t.Fatalf("failed to read fixture file %s: %v", tc.relPath, err) + } + + var gov decision.Governed + if err := json.Unmarshal(data, &gov); err != nil { + t.Fatalf("failed to unmarshal decision.Governed from %s: %v", tc.relPath, err) + } + + if err := gov.Decision.Auditable(); err != nil { + t.Fatalf("decision inside governed envelope fails Auditable invariant: %v", err) + } + + if diff := cmp.Diff(tc.wantSignalRef, gov.Decision.SignalRef); diff != "" { + t.Errorf("wrong decision SignalRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantVerdict, gov.Decision.Verdict); diff != "" { + t.Errorf("wrong decision Verdict (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantBand, gov.Decision.GrantedBand); diff != "" { + t.Errorf("wrong decision GrantedBand (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantPolicyVer, gov.Decision.PolicyVersion); diff != "" { + t.Errorf("wrong decision PolicyVersion (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantSignalRef, gov.Set.SignalRef); diff != "" { + t.Errorf("wrong set SignalRef in governed envelope (-want +got):\n%s", diff) + } + }) + } + }) + + t.Run("Telemetry fixtures unmarshal into valid metric and stream responses", func(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + relPath string + isProm bool + wantStatus string + wantResultType string + wantAppName string + }{ + "UnmarshalPrometheusTelemetry unmarshals prom-cart-failure JSON into valid vector response": { + relPath: filepath.Join("telemetry", "prom-cart-failure.json"), + isProm: true, + wantStatus: "success", + wantResultType: "vector", + wantAppName: "cart", + }, + "UnmarshalLokiTelemetry unmarshals loki-cart-failure JSON into valid streams response": { + relPath: filepath.Join("telemetry", "loki-cart-failure.json"), + isProm: false, + wantStatus: "success", + wantResultType: "streams", + wantAppName: "cartservice", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + data, err := os.ReadFile(tc.relPath) //nolint:gosec // G304: fixed testdata path, not user input + if err != nil { + t.Fatalf("failed to read fixture file %s: %v", tc.relPath, err) + } + + if tc.isProm { + var prom promResponse + if err := json.Unmarshal(data, &prom); err != nil { + t.Fatalf("failed to unmarshal Prometheus response from %s: %v", tc.relPath, err) + } + if diff := cmp.Diff(tc.wantStatus, prom.Status); diff != "" { + t.Errorf("wrong prom status (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantResultType, prom.Data.ResultType); diff != "" { + t.Errorf("wrong prom resultType (-want +got):\n%s", diff) + } + if len(prom.Data.Result) == 0 { + t.Fatal("want non-empty prom results, got empty") + } + if diff := cmp.Diff(tc.wantAppName, prom.Data.Result[0].Metric["app"]); diff != "" { + t.Errorf("wrong prom metric app label (-want +got):\n%s", diff) + } + } else { + var loki lokiResponse + if err := json.Unmarshal(data, &loki); err != nil { + t.Fatalf("failed to unmarshal Loki response from %s: %v", tc.relPath, err) + } + if diff := cmp.Diff(tc.wantStatus, loki.Status); diff != "" { + t.Errorf("wrong loki status (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantResultType, loki.Data.ResultType); diff != "" { + t.Errorf("wrong loki resultType (-want +got):\n%s", diff) + } + if len(loki.Data.Result) == 0 { + t.Fatal("want non-empty loki results, got empty") + } + if diff := cmp.Diff(tc.wantAppName, loki.Data.Result[0].Stream["app"]); diff != "" { + t.Errorf("wrong loki stream app label (-want +got):\n%s", diff) + } + } + }) + } + }) +} diff --git a/test/fixtures/proposals/cart-failure.json b/test/fixtures/proposals/cart-failure.json new file mode 100644 index 0000000..d768776 --- /dev/null +++ b/test/fixtures/proposals/cart-failure.json @@ -0,0 +1,230 @@ +{ + "name": "cart-burn-accel", + "runID": "slo_burn:cart/1786969860387330045", + "signalRef": "slo_burn:cart", + "sloRef": "cart-availability", + "saoSnapshot": { + "version": 1, + "assembledAt": "2026-08-02T11:46:10Z", + "signal": { + "confidence": 1, + "severity": { + "degradationPct": 0.5, + "trajectory": "accelerating" + }, + "blastRadius": { + "affectedPct": 0.12, + "velocity": "accelerating", + "downstreamConsumers": 2 + }, + "originService": "cart" + }, + "topology": { + "upstream": [ + { + "name": "flagd", + "state": "healthy", + "trafficShare": 0.03 + }, + { + "name": "redis", + "state": "healthy", + "trafficShare": 0.89 + }, + { + "name": "unknown", + "state": "degraded", + "trafficShare": 0.08 + } + ], + "downstream": [ + { + "name": "checkout", + "state": "degraded", + "trafficShare": 0.2 + }, + { + "name": "frontend", + "state": "healthy", + "trafficShare": 0.8 + } + ] + }, + "change": { + "events": [ + { + "id": "config/otel-demo/flagd-config/4283", + "type": "config", + "target": "flagd", + "age": 289392211635, + "historicalStaleness": 289392211635 + }, + { + "id": "deploy/otel-demo/checkout/1", + "type": "deploy", + "target": "checkout", + "age": 580392211635, + "historicalStaleness": 580392211635 + }, + { + "id": "deploy/otel-demo/cart/1", + "type": "deploy", + "target": "cart", + "age": 593392211635, + "historicalStaleness": 593392211635 + } + ] + } + }, + "failureClass": "service_failure", + "causalScores": [ + { + "eventID": "config/otel-demo/flagd-config/4283", + "inTopology": true, + "temporal": 0.89, + "historical": 0.9, + "liveCorroborated": true, + "likelihood": 0.85, + "rationale": [ + "temporal: 4m49s old -> 0.89", + "topological: in-path=true -> 0.20", + "historical: case-base prior -> 0.90" + ] + }, + { + "eventID": "deploy/otel-demo/checkout/1", + "inTopology": true, + "temporal": 0.8, + "topological": 0.2, + "historical": 0.9, + "liveCorroborated": true, + "likelihood": 0.63, + "rationale": [ + "temporal: 9m40s old -> 0.80", + "topological: in-path=true -> 0.20", + "historical: case-base prior -> 0.90" + ] + }, + { + "eventID": "deploy/otel-demo/cart/1", + "temporal": 0.8, + "historical": 0.9, + "likelihood": 0.5, + "rationale": [ + "temporal: 9m53s old -> 0.80", + "topological: in-path=false -> 0.00", + "historical: case-base prior -> 0.90", + "defence 1: uncorroborated -> capped at 0.50", + "target \"cart\" is not in this signal's topology -> scored for audit, excluded from confidence" + ] + } + ], + "hypotheses": [ + { + "name": "cartFailure flag injected via recent flagd config change", + "weight": 0.85 + }, + { + "name": "Cart pod crash loop or transient restart issue", + "weight": 0.1 + }, + { + "name": "Redis connectivity degradation affecting cart state operations", + "weight": 0.05 + } + ], + "evidence": [ + { + "tool": "metrics", + "query": "cart_error_ratio", + "key": "cart_error_ratio", + "summary": "cart_error_ratio = 0.50", + "ref": "metrics://cart_error_ratio", + "live": true, + "subject": "cart" + }, + { + "tool": "metrics", + "query": "severity_cart_availability", + "key": "severity_cart_availability", + "summary": "severity_cart_availability = 0.50", + "ref": "metrics://severity_cart_availability", + "live": true, + "subject": "cart" + }, + { + "tool": "kube", + "query": "{\"resource\":\"pods\",\"namespace\":\"otel-demo\",\"selector\":{\"app.kubernetes.io/name\":\"cart\"}}", + "key": "kube-1", + "summary": "cart-79bf56b7f9-r5ngh (Running)", + "ref": "kube://otel-demo/pods", + "live": true, + "subject": "cart" + }, + { + "tool": "loki", + "query": "{namespace=\"otel-demo\", service_name=\"cart\"}", + "key": "loki-1", + "summary": "50 log line(s); last: rpc error: code = Unavailable desc = connection error", + "ref": "loki://otel-demo/service_name=cart", + "live": true, + "subject": "cart" + } + ], + "serviceTier": "tier-1", + "gate": { + "budgetOK": true, + "dedupeOK": true, + "evidenceOK": true, + "passed": true + }, + "proposals": [ + { + "id": "p1", + "contractRef": "disable-cart-failure", + "confidence": 0.85, + "computedConfidence": 1, + "confidenceCeilingBound": true, + "terms": { + "signalConfidence": 1, + "corroborated": 3, + "grounding": 1, + "alignmentOK": true, + "likelihoodOK": true + }, + "predictedImpact": { + "severityReductionPct": 0.9, + "blastRadiusDelta": 0, + "sloEffects": { + "time_to_effect": "30s" + } + }, + "blastTier": "low", + "reversalPath": { + "method": "enable-cart-failure", + "watching": "cart_error_ratio", + "trigger": "cart_error_ratio < 0.02", + "automatic": true + }, + "governanceLevel": { + "band": "act_reversible" + }, + "rank": 1, + "citations": [ + "cart_error_ratio", + "severity_cart_availability", + "kube-1", + "loki-1" + ] + } + ], + "recommended": "p1", + "rankingRationale": { + "dominantAxis": "time_to_effect", + "velocityWeight": "accelerating" + }, + "status": { + "phase": "proposed", + "observedAt": "2026-08-02T11:46:30Z" + } +} diff --git a/test/fixtures/telemetry/loki-cart-failure.json b/test/fixtures/telemetry/loki-cart-failure.json new file mode 100644 index 0000000..f97a32a --- /dev/null +++ b/test/fixtures/telemetry/loki-cart-failure.json @@ -0,0 +1,20 @@ +{ + "status": "success", + "data": { + "resultType": "streams", + "result": [ + { + "stream": { + "app": "cartservice", + "namespace": "otel-demo" + }, + "values": [ + [ + "1750000000000000000", + "rpc error: code = Unavailable desc = connection error" + ] + ] + } + ] + } +} diff --git a/test/fixtures/telemetry/prom-cart-failure.json b/test/fixtures/telemetry/prom-cart-failure.json new file mode 100644 index 0000000..bf2f785 --- /dev/null +++ b/test/fixtures/telemetry/prom-cart-failure.json @@ -0,0 +1,18 @@ +{ + "status": "success", + "data": { + "resultType": "vector", + "result": [ + { + "metric": { + "app": "cart", + "job": "otel-demo" + }, + "value": [ + 1750000000, + "1.25" + ] + } + ] + } +} From 5fe1ea00e5ae8945ed9100a172a1cf8cacde02e3 Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:16:47 -0400 Subject: [PATCH 2/6] feat(step): implement isolated beat execution functions --- internal/step/export_test.go | 20 ++ internal/step/step.go | 334 ++++++++++++++++++ internal/step/step_test.go | 639 +++++++++++++++++++++++++++++++++++ 3 files changed, 993 insertions(+) create mode 100644 internal/step/export_test.go create mode 100644 internal/step/step.go create mode 100644 internal/step/step_test.go diff --git a/internal/step/export_test.go b/internal/step/export_test.go new file mode 100644 index 0000000..02fe91d --- /dev/null +++ b/internal/step/export_test.go @@ -0,0 +1,20 @@ +package step + +import ( + "context" + + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/internal/reason" +) + +// RunClankWithModelForTest allows tests to drive RunClank using a custom +// reason.Model seam rather than calling live Anthropic APIs. +func RunClankWithModelForTest(ctx context.Context, detectionFile, profileDir string, model reason.Model) (proposal.Set, error) { + return runClank(ctx, detectionFile, profileDir, model) +} + +// RunClankWithModelAndToolsForTest allows tests to drive RunClank using a +// custom reason.Model seam and injected tools. +func RunClankWithModelAndToolsForTest(ctx context.Context, detectionFile, profileDir string, model reason.Model, tools map[string]reason.Tool) (proposal.Set, error) { + return runClankWithTools(ctx, detectionFile, profileDir, model, tools) +} diff --git a/internal/step/step.go b/internal/step/step.go new file mode 100644 index 0000000..0caf154 --- /dev/null +++ b/internal/step/step.go @@ -0,0 +1,334 @@ +// Package step is the isolated beat execution plane: pure entrypoints that +// run each beat on static files without background daemons or cluster +// connectivity. +package step + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "sigs.k8s.io/yaml" + + sdk_anthropic "github.com/anthropics/anthropic-sdk-go" + + "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/outcome" + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/api/v1/signal" + "github.com/ianeff/thump/internal/actuate" + "github.com/ianeff/thump/internal/anthropic" + "github.com/ianeff/thump/internal/clank" + "github.com/ianeff/thump/internal/contract" + "github.com/ianeff/thump/internal/evidence" + "github.com/ianeff/thump/internal/hiss" + "github.com/ianeff/thump/internal/rattle" + "github.com/ianeff/thump/internal/reason" + "github.com/ianeff/thump/internal/thump" +) + +// ErrNoDetection indicates that a rattle reconciliation pass completed +// without triggering any detection rules. +var ErrNoDetection = errors.New("no signal detected") + +// RunHiss evaluates one proposal against policy and returns the governed +// decision envelope — hiss's pure authority pass over static files. +func RunHiss(ctx context.Context, proposalFile, policyFile string) (decision.Governed, error) { + if ctx.Err() != nil { + return decision.Governed{}, ctx.Err() + } + if proposalFile == "" { + return decision.Governed{}, errors.New("proposal file is required") + } + if policyFile == "" { + return decision.Governed{}, errors.New("policy file is required") + } + + data, err := os.ReadFile(proposalFile) //nolint:gosec // G304: operator-supplied proposal file path, not user input + if err != nil { + return decision.Governed{}, fmt.Errorf("read proposal file: %w", err) + } + + var ps proposal.Set + if err := yaml.Unmarshal(data, &ps); err != nil { + return decision.Governed{}, fmt.Errorf("parse proposal file: %w", err) + } + + pol, err := hiss.LoadPolicy(policyFile) + if err != nil { + return decision.Governed{}, fmt.Errorf("load policy: %w", err) + } + + auth := hiss.Authority{} + d := auth.Evaluate(ps, pol, time.Now()) + + return decision.Governed{Decision: d, Set: ps}, nil +} + +// RunThump renders and optionally executes one governed decision against the +// action catalog — dry-run by default, never touching infrastructure +// unattended. +func RunThump(ctx context.Context, decisionFile, catalogFile string, dryRun bool) (outcome.Outcome, error) { + if ctx.Err() != nil { + return outcome.Outcome{}, ctx.Err() + } + if decisionFile == "" { + return outcome.Outcome{}, errors.New("decision file is required") + } + if catalogFile == "" { + return outcome.Outcome{}, errors.New("catalog file is required") + } + + data, err := os.ReadFile(decisionFile) //nolint:gosec // G304: operator-supplied decision file path, not user input + if err != nil { + return outcome.Outcome{}, fmt.Errorf("read decision file: %w", err) + } + + var g decision.Governed + if err := yaml.Unmarshal(data, &g); err != nil { + return outcome.Outcome{}, fmt.Errorf("parse decision file: %w", err) + } + + cat, err := contract.LoadCatalogFile(catalogFile, contract.Preconditions) + if err != nil { + return outcome.Outcome{}, fmt.Errorf("load action catalog: %w", err) + } + + order, err := (thump.Actuator{}).Render(g, cat, time.Now()) + if err != nil { + return outcome.Outcome{}, fmt.Errorf("render order: %w", err) + } + + if dryRun { + exec := thump.DryRun{} + return exec.Execute(ctx, order, time.Now()), nil + } + + runner, err := actuate.New(cat, nil) + if err != nil { + return outcome.Outcome{}, fmt.Errorf("build live executor: %w", err) + } + exec := thump.Live{Runner: runner} + return exec.Execute(ctx, order, time.Now()), nil +} + +// RunClank reasons over a detection to produce an evidence-backed proposal +// set — the autonomous loop bounded by the action catalog. +func RunClank(ctx context.Context, detectionFile, profileDir, modelName, apiKey string) (proposal.Set, error) { + if ctx.Err() != nil { + return proposal.Set{}, ctx.Err() + } + if apiKey == "" { + apiKey = os.Getenv("ANTHROPIC_API_KEY") + } + if apiKey == "" { + return proposal.Set{}, errors.New("anthropic API key is required") + } + var m sdk_anthropic.Model + switch modelName { + case "", "haiku": + m = anthropic.ModelClaudeHaiku4_5 + case "sonnet": + m = anthropic.ModelClaudeSonnet5 + default: + m = sdk_anthropic.Model(modelName) + } + model := anthropic.NewModel(apiKey, m, 120*time.Second) + return runClank(ctx, detectionFile, profileDir, model) +} + +func runClank(ctx context.Context, detectionFile, profileDir string, model reason.Model) (proposal.Set, error) { + return runClankWithTools(ctx, detectionFile, profileDir, model, nil) +} + +func runClankWithTools(ctx context.Context, detectionFile, profileDir string, model reason.Model, overrideTools map[string]reason.Tool) (proposal.Set, error) { + if ctx.Err() != nil { + return proposal.Set{}, ctx.Err() + } + if detectionFile == "" { + return proposal.Set{}, errors.New("detection file is required") + } + if profileDir == "" { + return proposal.Set{}, errors.New("profile directory is required") + } + + data, err := os.ReadFile(detectionFile) //nolint:gosec // G304: operator-supplied detection file path, not user input + if err != nil { + return proposal.Set{}, fmt.Errorf("read detection file: %w", err) + } + + var det signal.Detection + if err := yaml.Unmarshal(data, &det); err != nil { + return proposal.Set{}, fmt.Errorf("parse detection file: %w", err) + } + + catPath, err := findFile( + filepath.Join(profileDir, "actions", "catalog.yaml"), + filepath.Join(profileDir, "catalog.yaml"), + ) + if err != nil { + return proposal.Set{}, fmt.Errorf("locate catalog: %w", err) + } + cat, err := contract.LoadCatalogFile(catPath, contract.Preconditions) + if err != nil { + return proposal.Set{}, fmt.Errorf("load action catalog: %w", err) + } + + classesPath, err := findFile( + filepath.Join(profileDir, "actions", "failure-classes.yaml"), + filepath.Join(profileDir, "failure-classes.yaml"), + ) + if err != nil { + return proposal.Set{}, fmt.Errorf("locate failure classes: %w", err) + } + classes, err := contract.LoadFailureClassesFile(classesPath) + if err != nil { + return proposal.Set{}, fmt.Errorf("load failure classes: %w", err) + } + + weightsPath, err := findFile( + filepath.Join(profileDir, "clank", "weights.yaml"), + filepath.Join(profileDir, "weights.yaml"), + filepath.Join(profileDir, "..", "clank", "weights.yaml"), + ) + if err != nil { + return proposal.Set{}, fmt.Errorf("locate weights: %w", err) + } + weights, err := clank.LoadWeightsFile(weightsPath) + if err != nil { + return proposal.Set{}, fmt.Errorf("load weights: %w", err) + } + + limitsPath, err := findFile( + filepath.Join(profileDir, "clank", "limits.yaml"), + filepath.Join(profileDir, "limits.yaml"), + filepath.Join(profileDir, "..", "clank", "limits.yaml"), + ) + if err != nil { + return proposal.Set{}, fmt.Errorf("locate limits: %w", err) + } + limits, err := clank.LoadLimitsFile(limitsPath) + if err != nil { + return proposal.Set{}, fmt.Errorf("load limits: %w", err) + } + + tools := overrideTools + if tools == nil { + tools = make(map[string]reason.Tool) + evPath, err := findFile( + filepath.Join(profileDir, "whir", "evidence-queries.yaml"), + filepath.Join(profileDir, "evidence-queries.yaml"), + ) + if err == nil { + if ev, err := evidence.LoadEvidenceConfig(evPath); err == nil { + if promURL := os.Getenv("PROM_URL"); promURL != "" && len(ev.Queries) > 0 { + tools["metrics"] = &evidence.MetricsTool{BaseURL: promURL, Queries: ev.Queries} + } + if lokiURL := os.Getenv("LOKI_URL"); lokiURL != "" { + tools["loki"] = &evidence.LokiTool{BaseURL: lokiURL, Subjects: ev.Index} + } + } + } + } + + ledger := clank.NewMemProposalLog() + ledger.LedgerRetention = limits.LedgerRetention + cases := clank.NewCaseBase() + cases.MaxCases = limits.MaxCases + + eng := &clank.Engine{ + Intake: clank.NewIntake(noopTopology{}, noopChange{}), + Model: model, + Tools: tools, + Catalog: cat, + FailureClasses: classes, + Ranker: clank.NewRanker(), + Store: clank.NewMemStore(), + Scorer: &clank.CausalScorerImpl{Prior: cases}, + Prior: cases, + Ledger: ledger, + Gate: clank.ReadinessGate{}, + MaxSteps: limits.MaxSteps, + Weights: weights, + } + + return eng.Propose(ctx, det) +} + +func findFile(paths ...string) (string, error) { + for _, p := range paths { + if _, err := os.Stat(p); err == nil { + return p, nil + } + } + return "", fmt.Errorf("file not found in candidates: %v", paths) +} + +type noopTopology struct{} + +func (noopTopology) Topology(_ context.Context, _ signal.Detection) (proposal.TopologySnapshot, error) { + return proposal.TopologySnapshot{}, nil +} + +type noopChange struct{} + +func (noopChange) Changes(_ context.Context, _ signal.Detection) (proposal.ChangeSnapshot, error) { + return proposal.ChangeSnapshot{}, nil +} + +// RunRattle reconciles configured SLOs against Prometheus telemetry to yield +// the first fired signal detection — rattle's evaluation pass in isolation. +func RunRattle(ctx context.Context, watchFile, queryConfigFile, promURL string) (signal.Detection, error) { + if ctx.Err() != nil { + return signal.Detection{}, ctx.Err() + } + if watchFile == "" { + return signal.Detection{}, errors.New("watch file is required") + } + if queryConfigFile == "" { + return signal.Detection{}, errors.New("query config file is required") + } + if promURL == "" { + return signal.Detection{}, errors.New("promURL is required") + } + + slos, err := rattle.LoadWatch(watchFile) + if err != nil { + return signal.Detection{}, fmt.Errorf("load watch file: %w", err) + } + + query, err := rattle.LoadQueryConfig(queryConfigFile) + if err != nil { + return signal.Detection{}, fmt.Errorf("load query config: %w", err) + } + + src := rattle.NewPromSource(promURL) + src.Step = query.Step + src.Window = query.Window + + r := &rattle.Reconciler{ + SLOs: slos, + Source: src, + Detector: rattle.AccelerationDetector{Threshold: 0.5}, + Sustained: &rattle.SustainedBurnDetector{Threshold: 1.0, MinSamples: query.SustainedMinSamples}, + Debounce: rattle.NewDebouncer(query.Debounce), + Contract: &rattle.SignalContract{ + FreshnessBound: query.FreshnessBound, + ConfidenceFloor: 0.5, + }, + } + + detections, err := r.Reconcile(ctx) + if err != nil { + return signal.Detection{}, fmt.Errorf("reconcile: %w", err) + } + + if len(detections) == 0 { + return signal.Detection{}, ErrNoDetection + } + + return detections[0], nil +} diff --git a/internal/step/step_test.go b/internal/step/step_test.go new file mode 100644 index 0000000..96fa9f0 --- /dev/null +++ b/internal/step/step_test.go @@ -0,0 +1,639 @@ +package step_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + + "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/outcome" + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/internal/reason" + "github.com/ianeff/thump/internal/step" + "github.com/ianeff/thump/internal/thump" +) + +type fakeModel struct { + script []reason.Completion + calls int +} + +func (f *fakeModel) Complete(_ context.Context, _ []reason.Message, _ []reason.ToolSpec) (reason.Completion, error) { + if f.calls >= len(f.script) { + return reason.Completion{}, errors.New("fake model exhausted script") + } + resp := f.script[f.calls] + f.calls++ + return resp, nil +} + +type fakeTool struct { + key string + summary string + live bool + subject string +} + +func (f *fakeTool) Name() string { return "metrics" } + +func (f *fakeTool) Spec() reason.ToolSpec { + return reason.ToolSpec{ + Name: "metrics", + Description: "queries metrics", + InputSchema: json.RawMessage(`{"type":"object"}`), + } +} + +func (f *fakeTool) Run(_ context.Context, _ json.RawMessage) (proposal.EvidenceRef, error) { + return proposal.EvidenceRef{ + Tool: "metrics", + Key: f.key, + Summary: f.summary, + Live: f.live, + Subject: f.subject, + }, nil +} + +func proposeArgs(t *testing.T, set proposal.Set) json.RawMessage { + t.Helper() + raw, err := json.Marshal(set) + if err != nil { + t.Fatalf("marshal propose args: %v", err) + } + return raw +} + +func TestRunHiss(t *testing.T) { + t.Parallel() + + validProposal := filepath.Join("..", "..", "test", "fixtures", "proposals", "cart-failure.json") + validPolicy := filepath.Join("..", "..", "config", "dev", "hiss", "policy.yaml") + + lowConfidenceProposal := filepath.Join(t.TempDir(), "low-confidence.json") + lowConfSet := proposal.Set{ + Name: "cart-burn-accel", + SignalRef: "slo_burn:cart", + SLORef: "cart-availability", + FailureClass: "service_failure", + ServiceTier: "tier-1", + Gate: &proposal.GateResult{ + BudgetOK: true, + DedupeOK: true, + EvidenceOK: true, + Passed: true, + }, + Proposals: []proposal.Candidate{ + { + ID: "p1", + ContractRef: "disable-cart-failure", + Confidence: 0.50, // below 0.75 floor for tier-1 service_failure + GovernanceLevel: &proposal.GovernanceLevel{ + Band: "act_reversible", + }, + ReversalPath: &proposal.ReversalPath{ + Method: "enable-cart-failure", + Watching: "cart_error_ratio", + Trigger: "cart_error_ratio < 0.02", + Automatic: true, + }, + }, + }, + Recommended: "p1", + } + lowConfRaw, err := json.Marshal(lowConfSet) + if err != nil { + t.Fatalf("marshal low confidence proposal: %v", err) + } + if err := os.WriteFile(lowConfidenceProposal, lowConfRaw, 0o600); err != nil { + t.Fatalf("write low confidence proposal: %v", err) + } + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + tests := map[string]struct { + ctx context.Context + proposalFile string + policyFile string + wantVerdict decision.Verdict + wantErr bool + wantTarget error + }{ + "RunHiss evaluates valid proposal and policy into approved governed decision": { + ctx: context.Background(), + proposalFile: validProposal, + policyFile: validPolicy, + wantVerdict: decision.VerdictApproved, + }, + "RunHiss escalates proposal when confidence is below policy floor": { + ctx: context.Background(), + proposalFile: lowConfidenceProposal, + policyFile: validPolicy, + wantVerdict: decision.VerdictEscalate, + }, + "RunHiss returns error for non-existent proposal file": { + ctx: context.Background(), + proposalFile: "nonexistent.json", + policyFile: validPolicy, + wantErr: true, + }, + "RunHiss returns error for non-existent policy file": { + ctx: context.Background(), + proposalFile: validProposal, + policyFile: "nonexistent.yaml", + wantErr: true, + }, + "RunHiss returns error when context is cancelled": { + ctx: canceledCtx, + proposalFile: validProposal, + policyFile: validPolicy, + wantErr: true, + wantTarget: context.Canceled, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + got, err := step.RunHiss(tc.ctx, tc.proposalFile, tc.policyFile) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.wantTarget != nil && !errors.Is(err, tc.wantTarget) { + t.Fatalf("expected error wrapping %v, got %v", tc.wantTarget, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if diff := cmp.Diff(tc.wantVerdict, got.Decision.Verdict); diff != "" { + t.Errorf("wrong decision verdict (-want +got):\n%s", diff) + } + if got.Decision.Verdict == decision.VerdictApproved { + if diff := cmp.Diff("slo_burn:cart", got.Decision.SignalRef); diff != "" { + t.Errorf("wrong signalRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff("p1", got.Decision.CandidateRef); diff != "" { + t.Errorf("wrong candidateRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(decision.BandActReversible, got.Decision.GrantedBand); diff != "" { + t.Errorf("wrong grantedBand (-want +got):\n%s", diff) + } + } + }) + } +} + +func TestRunThump(t *testing.T) { + t.Parallel() + + validDecision := filepath.Join("..", "..", "test", "fixtures", "decisions", "cart-failure.json") + validCatalog := filepath.Join("..", "..", "config", "dev", "actions", "catalog.yaml") + + escalateDecision := filepath.Join(t.TempDir(), "escalate-decision.json") + escGoverned := decision.Governed{ + Decision: decision.Decision{ + ID: "dec:slo_burn:cart:12345", + ProposalRef: "cart-burn-accel", + SignalRef: "slo_burn:cart", + SLORef: "cart-availability", + CandidateRef: "p1", + Verdict: decision.VerdictEscalate, + Reasons: []string{decision.ReasonConfidenceFloor}, + PolicyVersion: "v1", + EvaluatedAt: time.Now().UTC(), + }, + Set: proposal.Set{ + Name: "cart-burn-accel", + SignalRef: "slo_burn:cart", + SLORef: "cart-availability", + FailureClass: "service_failure", + Proposals: []proposal.Candidate{ + { + ID: "p1", + ContractRef: "disable-cart-failure", + }, + }, + Recommended: "p1", + }, + } + escRaw, err := json.Marshal(escGoverned) + if err != nil { + t.Fatalf("marshal escalate decision: %v", err) + } + if err := os.WriteFile(escalateDecision, escRaw, 0o600); err != nil { + t.Fatalf("write escalate decision: %v", err) + } + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + tests := map[string]struct { + ctx context.Context + decisionFile string + catalogFile string + dryRun bool + wantMode outcome.Mode + wantResult outcome.Result + wantContract string + wantErr bool + wantTarget error + }{ + "RunThump renders and dry-run executes approved governed decision": { + ctx: context.Background(), + decisionFile: validDecision, + catalogFile: validCatalog, + dryRun: true, + wantMode: outcome.ModeDryRun, + wantResult: outcome.ResultRendered, + wantContract: "disable-cart-failure", + }, + "RunThump returns error when decision verdict is not approved": { + ctx: context.Background(), + decisionFile: escalateDecision, + catalogFile: validCatalog, + dryRun: true, + wantErr: true, + wantTarget: thump.ErrUngoverned, + }, + "RunThump returns error for non-existent decision file": { + ctx: context.Background(), + decisionFile: "nonexistent.json", + catalogFile: validCatalog, + dryRun: true, + wantErr: true, + }, + "RunThump returns error for non-existent catalog file": { + ctx: context.Background(), + decisionFile: validDecision, + catalogFile: "nonexistent.yaml", + dryRun: true, + wantErr: true, + }, + "RunThump returns error when context is cancelled": { + ctx: canceledCtx, + decisionFile: validDecision, + catalogFile: validCatalog, + dryRun: true, + wantErr: true, + wantTarget: context.Canceled, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + got, err := step.RunThump(tc.ctx, tc.decisionFile, tc.catalogFile, tc.dryRun) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.wantTarget != nil && !errors.Is(err, tc.wantTarget) { + t.Fatalf("expected error wrapping %v, got %v", tc.wantTarget, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if diff := cmp.Diff(tc.wantMode, got.Mode); diff != "" { + t.Errorf("wrong outcome mode (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantResult, got.Result); diff != "" { + t.Errorf("wrong outcome result (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantContract, got.ContractRef); diff != "" { + t.Errorf("wrong contractRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff("slo_burn:cart", got.SignalRef); diff != "" { + t.Errorf("wrong signalRef (-want +got):\n%s", diff) + } + }) + } +} + +func TestRunClank(t *testing.T) { + t.Parallel() + + validDetection := filepath.Join("..", "..", "test", "fixtures", "detections", "cart-failure.json") + validProfile := filepath.Join("..", "..", "config", "dev") + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + tests := map[string]struct { + ctx context.Context + detectionFile string + profileDir string + model reason.Model + tools map[string]reason.Tool + wantRec string + wantErr bool + wantTarget error + }{ + "RunClank reasons over cart failure detection with mock model and produces valid proposal set": { + ctx: context.Background(), + detectionFile: validDetection, + profileDir: validProfile, + model: &fakeModel{script: []reason.Completion{ + {ToolCalls: []reason.ToolCall{{ + Name: "metrics", + Args: json.RawMessage(`{"q":"cart_error_ratio"}`), + }}}, + {ToolCalls: []reason.ToolCall{{ + Name: "propose", + Args: proposeArgs(t, proposal.Set{ + FailureClass: "service_failure", + Hypotheses: []proposal.Hypothesis{ + {Name: "cart-service-failure", Weight: 0.9}, + }, + Proposals: []proposal.Candidate{ + { + ID: "p1", + ContractRef: "disable-cart-failure", + Confidence: 0.85, + Citations: []string{"cart_error_ratio"}, + GovernanceLevel: &proposal.GovernanceLevel{ + Band: "act_reversible", + }, + ReversalPath: &proposal.ReversalPath{ + Method: "enable-cart-failure", + Watching: "cart_error_ratio", + Trigger: "cart_error_ratio < 0.02", + Automatic: true, + }, + }, + }, + }), + }}}, + }}, + tools: map[string]reason.Tool{ + "metrics": &fakeTool{key: "cart_error_ratio", summary: "cart_error_ratio = 0.50", live: true, subject: "cart"}, + }, + wantRec: "p1", + }, + "RunClank returns error when detection file does not exist": { + ctx: context.Background(), + detectionFile: "nonexistent.json", + profileDir: validProfile, + model: &fakeModel{}, + wantErr: true, + }, + "RunClank returns error when profile directory does not exist": { + ctx: context.Background(), + detectionFile: validDetection, + profileDir: "nonexistent-dir", + model: &fakeModel{}, + wantErr: true, + }, + "RunClank returns error when context is cancelled": { + ctx: canceledCtx, + detectionFile: validDetection, + profileDir: validProfile, + model: &fakeModel{}, + wantErr: true, + wantTarget: context.Canceled, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + got, err := step.RunClankWithModelAndToolsForTest(tc.ctx, tc.detectionFile, tc.profileDir, tc.model, tc.tools) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.wantTarget != nil && !errors.Is(err, tc.wantTarget) { + t.Fatalf("expected error wrapping %v, got %v", tc.wantTarget, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Gate == nil || !got.Gate.Passed { + t.Errorf("expected gate to pass, got %+v", got.Gate) + } + if diff := cmp.Diff(tc.wantRec, got.Recommended); diff != "" { + t.Errorf("wrong recommended proposal (-want +got):\n%s", diff) + } + if len(got.Proposals) != 1 { + t.Errorf("expected 1 proposal, got %d", len(got.Proposals)) + } + }) + } +} + +func TestRunClank_MissingAPIKey(t *testing.T) { + validDetection := filepath.Join("..", "..", "test", "fixtures", "detections", "cart-failure.json") + validProfile := filepath.Join("..", "..", "config", "dev") + + t.Setenv("ANTHROPIC_API_KEY", "") + _, err := step.RunClank(context.Background(), validDetection, validProfile, "", "") + if err == nil { + t.Fatal("expected error for missing API key, got nil") + } +} + +func TestRunRattle(t *testing.T) { + t.Parallel() + + validWatch := filepath.Join("..", "..", "config", "dev", "rattle", "watch.yaml") + validQuery := filepath.Join("..", "..", "config", "dev", "rattle", "query.yaml") + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + tests := map[string]struct { + ctx context.Context + watchFile string + queryConfigFile string + serverCtor func(t *testing.T) *httptest.Server + promURLOverride *string + wantService string + wantSLORef string + wantDetector string + wantErr bool + wantTarget error + }{ + "RunRattle detects accelerating burn rate from Prometheus response and emits signal detection": { + ctx: context.Background(), + watchFile: validWatch, + queryConfigFile: validQuery, + serverCtor: func(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + now := time.Now().Unix() + query := r.URL.Query().Get("query") + if query == `slo:current_burn_rate:ratio{sloth_id="cart-availability"}` { + resp := fmt.Sprintf(`{ + "status": "success", + "data": { + "resultType": "matrix", + "result": [{ + "metric": {"__name__": "slo:current_burn_rate:ratio", "sloth_id": "cart-availability"}, + "values": [ + [%d, "1.0"], + [%d, "2.0"], + [%d, "4.0"], + [%d, "8.0"] + ] + }] + } + }`, now-30, now-20, now-10, now) + _, _ = w.Write([]byte(resp)) + return + } + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + return srv + }, + wantService: "cart", + wantSLORef: "cart-availability", + wantDetector: "burn_rate_acceleration", + }, + "RunRattle returns ErrNoDetection when metrics do not breach acceleration threshold": { + ctx: context.Background(), + watchFile: validWatch, + queryConfigFile: validQuery, + serverCtor: func(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + now := time.Now().Unix() + query := r.URL.Query().Get("query") + if query == `slo:current_burn_rate:ratio{sloth_id="cart-availability"}` { + resp := fmt.Sprintf(`{ + "status": "success", + "data": { + "resultType": "matrix", + "result": [{ + "metric": {"__name__": "slo:current_burn_rate:ratio", "sloth_id": "cart-availability"}, + "values": [ + [%d, "0.1"], + [%d, "0.1"], + [%d, "0.1"], + [%d, "0.1"] + ] + }] + } + }`, now-30, now-20, now-10, now) + _, _ = w.Write([]byte(resp)) + return + } + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + return srv + }, + wantErr: true, + wantTarget: step.ErrNoDetection, + }, + "RunRattle returns error when watch file does not exist": { + ctx: context.Background(), + watchFile: "nonexistent.yaml", + queryConfigFile: validQuery, + serverCtor: func(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + return srv + }, + wantErr: true, + }, + "RunRattle returns error when query config file does not exist": { + ctx: context.Background(), + watchFile: validWatch, + queryConfigFile: "nonexistent.yaml", + serverCtor: func(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + return srv + }, + wantErr: true, + }, + "RunRattle returns error when promURL is empty": { + ctx: context.Background(), + watchFile: validWatch, + queryConfigFile: validQuery, + promURLOverride: stringPtr(""), + wantErr: true, + }, + "RunRattle returns error when context is cancelled": { + ctx: canceledCtx, + watchFile: validWatch, + queryConfigFile: validQuery, + serverCtor: func(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + return srv + }, + wantErr: true, + wantTarget: context.Canceled, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + promURL := "" + if tc.promURLOverride != nil { + promURL = *tc.promURLOverride + } else if tc.serverCtor != nil { + srv := tc.serverCtor(t) + promURL = srv.URL + } + + got, err := step.RunRattle(tc.ctx, tc.watchFile, tc.queryConfigFile, promURL) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.wantTarget != nil && !errors.Is(err, tc.wantTarget) { + t.Fatalf("expected error wrapping %v, got %v", tc.wantTarget, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if diff := cmp.Diff(tc.wantService, got.OriginService); diff != "" { + t.Errorf("wrong origin service (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantSLORef, got.SLORef); diff != "" { + t.Errorf("wrong sloRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantDetector, got.DetectorType); diff != "" { + t.Errorf("wrong detector type (-want +got):\n%s", diff) + } + }) + } +} + +func stringPtr(s string) *string { + return &s +} From 9f834b256c6995f95991471e75b3f4281ebae4c4 Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:20:59 -0400 Subject: [PATCH 3/6] feat(pipeline): implement headless simulation pipeline --- internal/pipeline/export_test.go | 13 ++ internal/pipeline/pipeline.go | 135 ++++++++++++++ internal/pipeline/pipeline_test.go | 290 +++++++++++++++++++++++++++++ internal/step/export_test.go | 2 +- internal/step/step.go | 6 +- 5 files changed, 443 insertions(+), 3 deletions(-) create mode 100644 internal/pipeline/export_test.go create mode 100644 internal/pipeline/pipeline.go create mode 100644 internal/pipeline/pipeline_test.go diff --git a/internal/pipeline/export_test.go b/internal/pipeline/export_test.go new file mode 100644 index 0000000..71567ff --- /dev/null +++ b/internal/pipeline/export_test.go @@ -0,0 +1,13 @@ +package pipeline + +import ( + "context" + + "github.com/ianeff/thump/internal/reason" +) + +// RunWithModelForTest allows tests to drive Run using a custom reason.Model +// seam and injected tools rather than calling live Anthropic APIs. +func RunWithModelForTest(ctx context.Context, detectionFile, profileDir string, model reason.Model, tools map[string]reason.Tool) (Result, error) { + return runWithModelAndTools(ctx, detectionFile, profileDir, model, tools) +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go new file mode 100644 index 0000000..1bddec9 --- /dev/null +++ b/internal/pipeline/pipeline.go @@ -0,0 +1,135 @@ +// Package pipeline chains the reasoning, governance, and actuation beats +// in-memory — a headless incident simulation that executes end-to-end without +// background daemons. +package pipeline + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "sigs.k8s.io/yaml" + + "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/outcome" + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/api/v1/signal" + "github.com/ianeff/thump/internal/contract" + "github.com/ianeff/thump/internal/hiss" + "github.com/ianeff/thump/internal/reason" + "github.com/ianeff/thump/internal/step" + "github.com/ianeff/thump/internal/thump" +) + +// Result is the immutable end-to-end record of an incident simulation — +// captures every beat transition from initial detection to rendered actuation. +type Result struct { + Detection signal.Detection `json:"detection"` // initial incident signal loaded from disk + Proposal proposal.Set `json:"proposal"` // ranked proposals synthesized by reasoning + Decision decision.Governed `json:"decision"` // policy-evaluated governance envelope + Outcome outcome.Outcome `json:"outcome"` // dry-run actuation outcome + Duration time.Duration `json:"duration"` // total simulation execution elapsed time +} + +// Run executes an end-to-end incident simulation from a detection file through +// clank reasoning, hiss governance, and dry-run thump actuation — dry-run by +// default, never touching infrastructure unattended. +func Run(ctx context.Context, detectionFile, profileDir, modelName, apiKey string) (Result, error) { + clankFn := func(ctx context.Context, detFile, profDir string) (proposal.Set, error) { + return step.RunClank(ctx, detFile, profDir, modelName, apiKey) + } + return runPipeline(ctx, detectionFile, profileDir, clankFn) +} + +func runWithModelAndTools(ctx context.Context, detectionFile, profileDir string, model reason.Model, tools map[string]reason.Tool) (Result, error) { + clankFn := func(ctx context.Context, detFile, profDir string) (proposal.Set, error) { + return step.RunClankWithModelAndTools(ctx, detFile, profDir, model, tools) + } + return runPipeline(ctx, detectionFile, profileDir, clankFn) +} + +func runPipeline(ctx context.Context, detectionFile, profileDir string, clankFn func(context.Context, string, string) (proposal.Set, error)) (Result, error) { + start := time.Now() + + if ctx.Err() != nil { + return Result{}, ctx.Err() + } + if detectionFile == "" { + return Result{}, errors.New("detection file is required") + } + if profileDir == "" { + return Result{}, errors.New("profile directory is required") + } + + data, err := os.ReadFile(detectionFile) //nolint:gosec // G304: operator-supplied detection file path, not user input + if err != nil { + return Result{}, fmt.Errorf("read detection file: %w", err) + } + + var det signal.Detection + if err := yaml.Unmarshal(data, &det); err != nil { + return Result{}, fmt.Errorf("parse detection file: %w", err) + } + + ps, err := clankFn(ctx, detectionFile, profileDir) + if err != nil { + return Result{}, fmt.Errorf("run clank: %w", err) + } + + policyPath, err := findFile( + filepath.Join(profileDir, "hiss", "policy.yaml"), + filepath.Join(profileDir, "policy.yaml"), + filepath.Join(profileDir, "..", "hiss", "policy.yaml"), + ) + if err != nil { + return Result{}, fmt.Errorf("locate policy: %w", err) + } + pol, err := hiss.LoadPolicy(policyPath) + if err != nil { + return Result{}, fmt.Errorf("load policy: %w", err) + } + + auth := hiss.Authority{} + d := auth.Evaluate(ps, pol, time.Now()) + gov := decision.Governed{Decision: d, Set: ps} + + catPath, err := findFile( + filepath.Join(profileDir, "actions", "catalog.yaml"), + filepath.Join(profileDir, "catalog.yaml"), + ) + if err != nil { + return Result{}, fmt.Errorf("locate catalog: %w", err) + } + cat, err := contract.LoadCatalogFile(catPath, contract.Preconditions) + if err != nil { + return Result{}, fmt.Errorf("load action catalog: %w", err) + } + + order, err := (thump.Actuator{}).Render(gov, cat, time.Now()) + if err != nil { + return Result{}, fmt.Errorf("render order: %w", err) + } + + exec := thump.DryRun{} + out := exec.Execute(ctx, order, time.Now()) + + return Result{ + Detection: det, + Proposal: ps, + Decision: gov, + Outcome: out, + Duration: time.Since(start), + }, nil +} + +func findFile(paths ...string) (string, error) { + for _, p := range paths { + if _, err := os.Stat(p); err == nil { + return p, nil + } + } + return "", fmt.Errorf("file not found in candidates: %v", paths) +} diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go new file mode 100644 index 0000000..3c2596a --- /dev/null +++ b/internal/pipeline/pipeline_test.go @@ -0,0 +1,290 @@ +package pipeline_test + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/outcome" + "github.com/ianeff/thump/api/v1/proposal" + "github.com/ianeff/thump/internal/pipeline" + "github.com/ianeff/thump/internal/reason" + "github.com/ianeff/thump/internal/thump" +) + +type fakeModel struct { + script []reason.Completion + calls int +} + +func (f *fakeModel) Complete(_ context.Context, _ []reason.Message, _ []reason.ToolSpec) (reason.Completion, error) { + if f.calls >= len(f.script) { + return reason.Completion{}, errors.New("fake model exhausted script") + } + resp := f.script[f.calls] + f.calls++ + return resp, nil +} + +type fakeTool struct { + name string + key string + summary string + live bool + subject string +} + +func (f *fakeTool) Name() string { + if f.name != "" { + return f.name + } + return "metrics" +} + +func (f *fakeTool) Spec() reason.ToolSpec { + return reason.ToolSpec{ + Name: f.Name(), + Description: "queries " + f.Name(), + InputSchema: json.RawMessage(`{"type":"object"}`), + } +} + +func (f *fakeTool) Run(_ context.Context, _ json.RawMessage) (proposal.EvidenceRef, error) { + return proposal.EvidenceRef{ + Tool: f.Name(), + Key: f.key, + Summary: f.summary, + Live: f.live, + Subject: f.subject, + }, nil +} + +func proposeArgs(t *testing.T, set proposal.Set) json.RawMessage { + t.Helper() + raw, err := json.Marshal(set) + if err != nil { + t.Fatalf("marshal propose args: %v", err) + } + return raw +} + +func TestRun(t *testing.T) { + t.Parallel() + + validDetection := filepath.Join("..", "..", "test", "fixtures", "detections", "cart-failure.json") + validProfile := filepath.Join("..", "..", "config", "dev") + + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + + validModel := &fakeModel{script: []reason.Completion{ + {ToolCalls: []reason.ToolCall{{ + Name: "metrics", + Args: json.RawMessage(`{"q":"cart_error_ratio"}`), + }}}, + {ToolCalls: []reason.ToolCall{{ + Name: "loki", + Args: json.RawMessage(`{"q":"cart_errors"}`), + }}}, + {ToolCalls: []reason.ToolCall{{ + Name: "propose", + Args: proposeArgs(t, proposal.Set{ + FailureClass: "service_failure", + Hypotheses: []proposal.Hypothesis{ + {Name: "cart-service-failure", Weight: 0.9}, + }, + Proposals: []proposal.Candidate{ + { + ID: "p1", + ContractRef: "disable-cart-failure", + Confidence: 0.85, + Citations: []string{"cart_error_ratio", "loki_cart_errors"}, + GovernanceLevel: &proposal.GovernanceLevel{ + Band: "act_reversible", + }, + ReversalPath: &proposal.ReversalPath{ + Method: "enable-cart-failure", + Watching: "cart_error_ratio", + Trigger: "cart_error_ratio < 0.02", + Automatic: true, + }, + }, + }, + }), + }}}, + }} + + lowConfidenceModel := &fakeModel{script: []reason.Completion{ + {ToolCalls: []reason.ToolCall{{ + Name: "metrics", + Args: json.RawMessage(`{"q":"cart_error_ratio"}`), + }}}, + {ToolCalls: []reason.ToolCall{{ + Name: "propose", + Args: proposeArgs(t, proposal.Set{ + FailureClass: "service_failure", + Hypotheses: []proposal.Hypothesis{ + {Name: "cart-service-failure", Weight: 0.9}, + }, + Proposals: []proposal.Candidate{ + { + ID: "p1", + ContractRef: "disable-cart-failure", + Confidence: 0.50, // below 0.75 floor for tier-1 service_failure + Citations: []string{"cart_error_ratio"}, + GovernanceLevel: &proposal.GovernanceLevel{ + Band: "act_reversible", + }, + ReversalPath: &proposal.ReversalPath{ + Method: "enable-cart-failure", + Watching: "cart_error_ratio", + Trigger: "cart_error_ratio < 0.02", + Automatic: true, + }, + }, + }, + }), + }}}, + }} + + validTools := map[string]reason.Tool{ + "metrics": &fakeTool{name: "metrics", key: "cart_error_ratio", summary: "cart_error_ratio = 0.50", live: true, subject: "cart"}, + "loki": &fakeTool{name: "loki", key: "loki_cart_errors", summary: "50 log lines with error", live: true, subject: "cart"}, + } + + tests := map[string]struct { + ctx context.Context + detectionFile string + profileDir string + model reason.Model + tools map[string]reason.Tool + wantService string + wantRec string + wantVerdict decision.Verdict + wantMode outcome.Mode + wantResult outcome.Result + wantContract string + wantErr bool + wantTarget error + }{ + "RunWithModelForTest executes cart failure simulation end-to-end and yields rendered dry-run outcome": { + ctx: context.Background(), + detectionFile: validDetection, + profileDir: validProfile, + model: validModel, + tools: validTools, + wantService: "cart", + wantRec: "p1", + wantVerdict: decision.VerdictApproved, + wantMode: outcome.ModeDryRun, + wantResult: outcome.ResultRendered, + wantContract: "disable-cart-failure", + }, + "RunWithModelForTest returns error when policy evaluation rejects or escalates decision": { + ctx: context.Background(), + detectionFile: validDetection, + profileDir: validProfile, + model: lowConfidenceModel, + tools: validTools, + wantErr: true, + wantTarget: thump.ErrUngoverned, + }, + "RunWithModelForTest returns error when detection file does not exist": { + ctx: context.Background(), + detectionFile: "nonexistent.json", + profileDir: validProfile, + model: &fakeModel{}, + wantErr: true, + }, + "RunWithModelForTest returns error when profile directory does not exist": { + ctx: context.Background(), + detectionFile: validDetection, + profileDir: "nonexistent-dir", + model: &fakeModel{}, + wantErr: true, + }, + "RunWithModelForTest returns error when context is cancelled": { + ctx: canceledCtx, + detectionFile: validDetection, + profileDir: validProfile, + model: &fakeModel{}, + wantErr: true, + wantTarget: context.Canceled, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + got, err := pipeline.RunWithModelForTest(tc.ctx, tc.detectionFile, tc.profileDir, tc.model, tc.tools) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tc.wantTarget != nil && !errors.Is(err, tc.wantTarget) { + t.Fatalf("expected error wrapping %v, got %v", tc.wantTarget, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if diff := cmp.Diff(tc.wantService, got.Detection.OriginService); diff != "" { + t.Errorf("wrong detection origin service (-want +got):\n%s", diff) + } + if diff := cmp.Diff("slo_burn:cart", got.Detection.Fingerprint); diff != "" { + t.Errorf("wrong detection fingerprint (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantRec, got.Proposal.Recommended); diff != "" { + t.Errorf("wrong proposal recommended (-want +got):\n%s", diff) + } + if got.Proposal.Gate == nil || !got.Proposal.Gate.Passed { + t.Errorf("expected proposal gate to pass, got %+v", got.Proposal.Gate) + } + if diff := cmp.Diff(tc.wantVerdict, got.Decision.Decision.Verdict); diff != "" { + t.Errorf("wrong decision verdict (-want +got):\n%s", diff) + } + if diff := cmp.Diff("slo_burn:cart", got.Decision.Decision.SignalRef); diff != "" { + t.Errorf("wrong decision signalRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantRec, got.Decision.Decision.CandidateRef); diff != "" { + t.Errorf("wrong decision candidateRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff(decision.BandActReversible, got.Decision.Decision.GrantedBand); diff != "" { + t.Errorf("wrong decision grantedBand (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantMode, got.Outcome.Mode); diff != "" { + t.Errorf("wrong outcome mode (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantResult, got.Outcome.Result); diff != "" { + t.Errorf("wrong outcome result (-want +got):\n%s", diff) + } + if diff := cmp.Diff(tc.wantContract, got.Outcome.ContractRef); diff != "" { + t.Errorf("wrong outcome contractRef (-want +got):\n%s", diff) + } + if diff := cmp.Diff("slo_burn:cart", got.Outcome.SignalRef); diff != "" { + t.Errorf("wrong outcome signalRef (-want +got):\n%s", diff) + } + if got.Duration <= 0 { + t.Errorf("expected positive duration, got %v", got.Duration) + } + }) + } +} + +func TestRun_MissingAPIKey(t *testing.T) { + validDetection := filepath.Join("..", "..", "test", "fixtures", "detections", "cart-failure.json") + validProfile := filepath.Join("..", "..", "config", "dev") + + t.Setenv("ANTHROPIC_API_KEY", "") + _, err := pipeline.Run(context.Background(), validDetection, validProfile, "", "") + if err == nil { + t.Fatal("expected error for missing API key, got nil") + } +} diff --git a/internal/step/export_test.go b/internal/step/export_test.go index 02fe91d..55c2ae9 100644 --- a/internal/step/export_test.go +++ b/internal/step/export_test.go @@ -16,5 +16,5 @@ func RunClankWithModelForTest(ctx context.Context, detectionFile, profileDir str // RunClankWithModelAndToolsForTest allows tests to drive RunClank using a // custom reason.Model seam and injected tools. func RunClankWithModelAndToolsForTest(ctx context.Context, detectionFile, profileDir string, model reason.Model, tools map[string]reason.Tool) (proposal.Set, error) { - return runClankWithTools(ctx, detectionFile, profileDir, model, tools) + return RunClankWithModelAndTools(ctx, detectionFile, profileDir, model, tools) } diff --git a/internal/step/step.go b/internal/step/step.go index 0caf154..67ab14a 100644 --- a/internal/step/step.go +++ b/internal/step/step.go @@ -141,10 +141,12 @@ func RunClank(ctx context.Context, detectionFile, profileDir, modelName, apiKey } func runClank(ctx context.Context, detectionFile, profileDir string, model reason.Model) (proposal.Set, error) { - return runClankWithTools(ctx, detectionFile, profileDir, model, nil) + return RunClankWithModelAndTools(ctx, detectionFile, profileDir, model, nil) } -func runClankWithTools(ctx context.Context, detectionFile, profileDir string, model reason.Model, overrideTools map[string]reason.Tool) (proposal.Set, error) { +// RunClankWithModelAndTools reasons over a detection using an explicit model +// and toolset — the test seam for hermetic reasoning without network access. +func RunClankWithModelAndTools(ctx context.Context, detectionFile, profileDir string, model reason.Model, overrideTools map[string]reason.Tool) (proposal.Set, error) { if ctx.Err() != nil { return proposal.Set{}, ctx.Err() } From 2fd0d59ff0cc140bee0ddd858d95604f5bca5e52 Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:24:02 -0400 Subject: [PATCH 4/6] feat(mock): implement telemetry and broker stub servers --- internal/mock/server.go | 335 ++++++++++++++++++++++++++++ internal/mock/server_test.go | 415 +++++++++++++++++++++++++++++++++++ 2 files changed, 750 insertions(+) create mode 100644 internal/mock/server.go create mode 100644 internal/mock/server_test.go diff --git a/internal/mock/server.go b/internal/mock/server.go new file mode 100644 index 0000000..0a3322c --- /dev/null +++ b/internal/mock/server.go @@ -0,0 +1,335 @@ +// Package mock provides in-process telemetry stub servers and embedded message +// brokers for zero-infrastructure simulation and testing — lightweight stubs +// execute locally with negligible resource overhead. +package mock + +import ( + "errors" + "fmt" + "net" + "net/http" + "os" + "time" + + natssrv "github.com/nats-io/nats-server/v2/server" +) + +// DefaultPrometheusFixture carries the default Prometheus instant and range +// query response — golden scenario telemetry from prom-cart-failure.json. +var DefaultPrometheusFixture = []byte(`{ + "status": "success", + "data": { + "resultType": "vector", + "result": [ + { + "metric": { + "app": "cart", + "job": "otel-demo" + }, + "value": [ + 1750000000, + "1.25" + ] + } + ] + } +}`) + +// DefaultLokiFixture carries the default Loki log query response — golden +// scenario telemetry from loki-cart-failure.json. +var DefaultLokiFixture = []byte(`{ + "status": "success", + "data": { + "resultType": "streams", + "result": [ + { + "stream": { + "app": "cartservice", + "namespace": "otel-demo" + }, + "values": [ + [ + "1750000000000000000", + "rpc error: code = Unavailable desc = connection error" + ] + ] + } + ] + } +}`) + +// TelemetryServer serves canned Prometheus and Loki HTTP telemetry endpoints — +// queries return configured fixtures or golden cart failure scenario defaults. +type TelemetryServer struct { + srv *http.Server + listener net.Listener + url string + promFixture []byte + lokiFixture []byte + promResponses map[string][]byte + lokiResponses map[string][]byte + port int +} + +// TelemetryOption configures a TelemetryServer during initialization. +type TelemetryOption func(*TelemetryServer) + +// WithPrometheusFixture overrides the default Prometheus JSON fixture served +// for instant and range queries. +func WithPrometheusFixture(fixture []byte) TelemetryOption { + return func(s *TelemetryServer) { + s.promFixture = fixture + } +} + +// WithLokiFixture overrides the default Loki JSON fixture served for range +// queries. +func WithLokiFixture(fixture []byte) TelemetryOption { + return func(s *TelemetryServer) { + s.lokiFixture = fixture + } +} + +// WithPrometheusResponses configures query-specific responses for Prometheus +// PromQL queries — unmatched queries fall back to the default fixture. +func WithPrometheusResponses(responses map[string][]byte) TelemetryOption { + return func(s *TelemetryServer) { + s.promResponses = responses + } +} + +// WithLokiResponses configures query-specific responses for Loki LogQL +// queries — unmatched queries fall back to the default fixture. +func WithLokiResponses(responses map[string][]byte) TelemetryOption { + return func(s *TelemetryServer) { + s.lokiResponses = responses + } +} + +// WithPort sets the TCP port for TelemetryServer — 0 selects a random +// ephemeral port. +func WithPort(port int) TelemetryOption { + return func(s *TelemetryServer) { + s.port = port + } +} + +// WithListener sets the net.Listener for TelemetryServer — overrides port +// binding configuration. +func WithListener(l net.Listener) TelemetryOption { + return func(s *TelemetryServer) { + s.listener = l + } +} + +// NewTelemetryServer initializes and starts an HTTP telemetry stub server — +// callers must invoke Close when finished to release listener resources. +func NewTelemetryServer(opts ...TelemetryOption) (*TelemetryServer, error) { + s := &TelemetryServer{ + promFixture: DefaultPrometheusFixture, + lokiFixture: DefaultLokiFixture, + } + for _, opt := range opts { + opt(s) + } + + mux := http.NewServeMux() + mux.HandleFunc("/ready", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ready"}`)) + }) + + handleProm := func(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query().Get("query") + if query == "" && r.Method == http.MethodPost { + _ = r.ParseForm() + query = r.Form.Get("query") + } + w.Header().Set("Content-Type", "application/json") + if resp, ok := s.promResponses[query]; ok { + _, _ = w.Write(resp) + return + } + _, _ = w.Write(s.promFixture) + } + mux.HandleFunc("/api/v1/query", handleProm) + mux.HandleFunc("/api/v1/query_range", handleProm) + + handleLoki := func(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query().Get("query") + if query == "" && r.Method == http.MethodPost { + _ = r.ParseForm() + query = r.Form.Get("query") + } + w.Header().Set("Content-Type", "application/json") + if resp, ok := s.lokiResponses[query]; ok { + _, _ = w.Write(resp) + return + } + _, _ = w.Write(s.lokiFixture) + } + mux.HandleFunc("/loki/api/v1/query_range", handleLoki) + + if s.listener == nil { + l, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", s.port)) + if err != nil { + return nil, fmt.Errorf("listen on 127.0.0.1:%d: %w", s.port, err) + } + s.listener = l + } + + s.url = fmt.Sprintf("http://%s", s.listener.Addr().String()) + s.srv = &http.Server{ + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, + } + + go func() { + _ = s.srv.Serve(s.listener) + }() + + return s, nil +} + +// URL returns the base HTTP URL of the running TelemetryServer. +func (s *TelemetryServer) URL() string { + return s.url +} + +// Addr returns the net.Addr of the underlying network listener. +func (s *TelemetryServer) Addr() net.Addr { + if s.listener == nil { + return nil + } + return s.listener.Addr() +} + +// Close stops the HTTP server and releases its network listener. +func (s *TelemetryServer) Close() error { + if s.srv == nil { + return nil + } + return s.srv.Close() +} + +// EmbeddedBroker runs an in-process NATS JetStream server for isolated tests +// and simulations — no external daemon or container is required. +type EmbeddedBroker struct { + srv *natssrv.Server + storeDir string + isTempStore bool +} + +type brokerConfig struct { + host string + port int + storeDir string +} + +// BrokerOption configures an EmbeddedBroker during initialization. +type BrokerOption func(*brokerConfig) + +// WithBrokerHost sets the listening host for EmbeddedBroker — defaults to +// 127.0.0.1. +func WithBrokerHost(host string) BrokerOption { + return func(c *brokerConfig) { + c.host = host + } +} + +// WithBrokerPort sets the listening port for EmbeddedBroker — defaults to +// -1 for an ephemeral port. +func WithBrokerPort(port int) BrokerOption { + return func(c *brokerConfig) { + c.port = port + } +} + +// WithBrokerStoreDir sets the JetStream storage directory — defaults to a +// temporary directory cleaned up on Close. +func WithBrokerStoreDir(dir string) BrokerOption { + return func(c *brokerConfig) { + c.storeDir = dir + } +} + +// NewEmbeddedBroker starts an in-process NATS JetStream server — callers must +// invoke Close when finished to terminate the server. +func NewEmbeddedBroker(opts ...BrokerOption) (*EmbeddedBroker, error) { + cfg := brokerConfig{ + host: "127.0.0.1", + port: -1, + } + for _, opt := range opts { + opt(&cfg) + } + + storeDir := cfg.storeDir + isTemp := false + if storeDir == "" { + tmp, err := os.MkdirTemp("", "thump-mock-broker-*") + if err != nil { + return nil, fmt.Errorf("create temporary jetstream store dir: %w", err) + } + storeDir = tmp + isTemp = true + } + + srv, err := natssrv.NewServer(&natssrv.Options{ + Host: cfg.host, + Port: cfg.port, + JetStream: true, + StoreDir: storeDir, + }) + if err != nil { + if isTemp { + _ = os.RemoveAll(storeDir) + } + return nil, fmt.Errorf("create embedded nats server: %w", err) + } + + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + srv.Shutdown() + if isTemp { + _ = os.RemoveAll(storeDir) + } + return nil, errors.New("embedded nats server not ready for connections") + } + + return &EmbeddedBroker{ + srv: srv, + storeDir: storeDir, + isTempStore: isTemp, + }, nil +} + +// ClientURL returns the NATS client connection URL. +func (b *EmbeddedBroker) ClientURL() string { + if b.srv == nil { + return "" + } + return b.srv.ClientURL() +} + +// Ready reports whether the broker is running and ready for client +// connections. +func (b *EmbeddedBroker) Ready() bool { + return b.srv != nil && b.srv.ReadyForConnections(100*time.Millisecond) +} + +// Close terminates the embedded NATS server and removes any temporary store +// directory. +func (b *EmbeddedBroker) Close() error { + if b.srv != nil { + b.srv.Shutdown() + b.srv.WaitForShutdown() + b.srv = nil + } + if b.isTempStore && b.storeDir != "" { + _ = os.RemoveAll(b.storeDir) + b.storeDir = "" + } + return nil +} diff --git a/internal/mock/server_test.go b/internal/mock/server_test.go new file mode 100644 index 0000000..172c806 --- /dev/null +++ b/internal/mock/server_test.go @@ -0,0 +1,415 @@ +package mock_test + +import ( + "context" + "encoding/json" + "io" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/ianeff/thump/internal/evidence" + "github.com/ianeff/thump/internal/httpx" + "github.com/ianeff/thump/internal/mock" +) + +func TestFixtures_MatchDiskGoldenFiles(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + diskPath string + defaultB []byte + }{ + "Prometheus default fixture matches disk fixture file": { + diskPath: filepath.Join("..", "..", "test", "fixtures", "telemetry", "prom-cart-failure.json"), + defaultB: mock.DefaultPrometheusFixture, + }, + "Loki default fixture matches disk fixture file": { + diskPath: filepath.Join("..", "..", "test", "fixtures", "telemetry", "loki-cart-failure.json"), + defaultB: mock.DefaultLokiFixture, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + diskBytes, err := os.ReadFile(tc.diskPath) //nolint:gosec // G304: fixed testdata path, not user input + if err != nil { + t.Fatalf("failed to read disk fixture %s: %v", tc.diskPath, err) + } + + var want any + if err := json.Unmarshal(diskBytes, &want); err != nil { + t.Fatalf("failed to unmarshal disk fixture %s: %v", tc.diskPath, err) + } + + var got any + if err := json.Unmarshal(tc.defaultB, &got); err != nil { + t.Fatalf("failed to unmarshal default fixture: %v", err) + } + + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("default fixture mismatch (-want +got):\n%s", diff) + } + }) + } +} + +func TestTelemetryServer_Defaults(t *testing.T) { + t.Parallel() + + srv, err := mock.NewTelemetryServer() + if err != nil { + t.Fatalf("NewTelemetryServer returned unexpected error: %v", err) + } + t.Cleanup(func() { + _ = srv.Close() + }) + + if srv.URL() == "" { + t.Fatal("srv.URL() is empty") + } + if srv.Addr() == nil { + t.Fatal("srv.Addr() is nil") + } + + client := &http.Client{Timeout: 5 * time.Second} + + tests := map[string]struct { + path string + method string + wantStatus int + wantJSONObj any + }{ + "Health check returns 200 OK with ready status": { + path: "/ready", + method: http.MethodGet, + wantStatus: http.StatusOK, + wantJSONObj: map[string]any{"status": "ready"}, + }, + "Prometheus instant query returns default cart failure fixture": { + path: "/api/v1/query?query=up", + method: http.MethodGet, + wantStatus: http.StatusOK, + wantJSONObj: unmarshalJSON(t, mock.DefaultPrometheusFixture), + }, + "Prometheus range query returns default cart failure fixture": { + path: "/api/v1/query_range?query=rate(http_requests_total[5m])", + method: http.MethodGet, + wantStatus: http.StatusOK, + wantJSONObj: unmarshalJSON(t, mock.DefaultPrometheusFixture), + }, + "Loki range query returns default cart failure fixture": { + path: "/loki/api/v1/query_range?query={app=\"cart\"}", + method: http.MethodGet, + wantStatus: http.StatusOK, + wantJSONObj: unmarshalJSON(t, mock.DefaultLokiFixture), + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + req, err := http.NewRequestWithContext(context.Background(), tc.method, srv.URL()+tc.path, nil) + if err != nil { + t.Fatalf("failed to create request: %v", err) + } + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected request error: %v", err) + } + defer func() { + if closeErr := resp.Body.Close(); closeErr != nil { + t.Errorf("error closing response body: %v", closeErr) + } + }() + + if resp.StatusCode != tc.wantStatus { + t.Errorf("status code mismatch: want %d, got %d", tc.wantStatus, resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("failed to read response body: %v", err) + } + + var gotJSON any + if err := json.Unmarshal(body, &gotJSON); err != nil { + t.Fatalf("failed to unmarshal body JSON: %v", err) + } + + if diff := cmp.Diff(tc.wantJSONObj, gotJSON); diff != "" { + t.Errorf("response body mismatch (-want +got):\n%s", diff) + } + }) + } +} + +func TestTelemetryServer_CustomFixturesAndResponses(t *testing.T) { + t.Parallel() + + customProm := []byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"app":"payment"},"value":[100,"42.0"]}]}}`) + customLoki := []byte(`{"status":"success","data":{"resultType":"streams","result":[{"stream":{"app":"payment"},"values":[["100","payment failed"]]}]}}`) + overrideProm := []byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"app":"override"},"value":[200,"99.9"]}]}}`) + overrideLoki := []byte(`{"status":"success","data":{"resultType":"streams","result":[{"stream":{"app":"override"},"values":[["200","override log"]]}]}}`) + + srv, err := mock.NewTelemetryServer( + mock.WithPrometheusFixture(customProm), + mock.WithLokiFixture(customLoki), + mock.WithPrometheusResponses(map[string][]byte{ + "custom_metric_query": overrideProm, + }), + mock.WithLokiResponses(map[string][]byte{ + `{app="custom"}`: overrideLoki, + }), + ) + if err != nil { + t.Fatalf("NewTelemetryServer returned unexpected error: %v", err) + } + t.Cleanup(func() { + _ = srv.Close() + }) + + client := &http.Client{Timeout: 5 * time.Second} + + tests := map[string]struct { + path string + wantJSONObj any + }{ + "Prometheus instant query returns custom fixture when unmapped": { + path: "/api/v1/query?query=fallback_query", + wantJSONObj: unmarshalJSON(t, customProm), + }, + "Prometheus instant query returns mapped response when query matches exact expression": { + path: "/api/v1/query?query=custom_metric_query", + wantJSONObj: unmarshalJSON(t, overrideProm), + }, + "Prometheus range query returns mapped response when query matches exact expression": { + path: "/api/v1/query_range?query=custom_metric_query", + wantJSONObj: unmarshalJSON(t, overrideProm), + }, + "Loki query returns custom fixture when unmapped": { + path: "/loki/api/v1/query_range?query={app=\"other\"}", + wantJSONObj: unmarshalJSON(t, customLoki), + }, + "Loki query returns mapped response when query matches exact expression": { + path: "/loki/api/v1/query_range?query={app=\"custom\"}", + wantJSONObj: unmarshalJSON(t, overrideLoki), + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL()+tc.path, nil) + if err != nil { + t.Fatalf("failed to create request: %v", err) + } + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected request error: %v", err) + } + defer func() { + if closeErr := resp.Body.Close(); closeErr != nil { + t.Errorf("error closing response body: %v", closeErr) + } + }() + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("failed to read response body: %v", err) + } + + var gotJSON any + if err := json.Unmarshal(body, &gotJSON); err != nil { + t.Fatalf("failed to unmarshal body JSON: %v", err) + } + + if diff := cmp.Diff(tc.wantJSONObj, gotJSON); diff != "" { + t.Errorf("response body mismatch (-want +got):\n%s", diff) + } + }) + } +} + +func TestTelemetryServer_Close(t *testing.T) { + t.Parallel() + + srv, err := mock.NewTelemetryServer() + if err != nil { + t.Fatalf("NewTelemetryServer returned unexpected error: %v", err) + } + + serverURL := srv.URL() + if err := srv.Close(); err != nil { + t.Fatalf("srv.Close() returned error: %v", err) + } + + client := &http.Client{Timeout: 500 * time.Millisecond} + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, serverURL+"/ready", nil) + if err != nil { + t.Fatalf("failed to create request: %v", err) + } + + resp, err := client.Do(req) + if err == nil { + _ = resp.Body.Close() + t.Fatal("expected error connecting to closed server, got nil") + } +} + +func TestTelemetryServer_Integration(t *testing.T) { + t.Parallel() + + srv, err := mock.NewTelemetryServer() + if err != nil { + t.Fatalf("NewTelemetryServer returned unexpected error: %v", err) + } + t.Cleanup(func() { + _ = srv.Close() + }) + + t.Run("httpx InstantQuery decodes vector result from mock telemetry server", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + result, err := httpx.InstantQuery(ctx, nil, srv.URL(), "cart_failure_rate") + if err != nil { + t.Fatalf("InstantQuery failed: %v", err) + } + + if len(result.Data.Result) != 1 { + t.Fatalf("expected 1 result, got %d", len(result.Data.Result)) + } + + metricApp := result.Data.Result[0].Metric["app"] + if metricApp != "cart" { + t.Errorf("metric app mismatch: want cart, got %s", metricApp) + } + + var val string + if err := json.Unmarshal(result.Data.Result[0].Value[1], &val); err != nil { + t.Fatalf("failed to unmarshal metric value: %v", err) + } + if val != "1.25" { + t.Errorf("metric value mismatch: want 1.25, got %s", val) + } + }) + + t.Run("evidence LokiTool receives live logs from mock telemetry server", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + tool := &evidence.LokiTool{ + BaseURL: srv.URL(), + } + + ref, err := tool.Run(ctx, json.RawMessage(`{"namespace":"otel-demo","labels":{"app":"cartservice"}}`)) + if err != nil { + t.Fatalf("LokiTool.Run failed: %v", err) + } + + if !ref.Live { + t.Errorf("expected ref.Live to be true, got false (summary: %s)", ref.Summary) + } + }) +} + +func TestEmbeddedBroker_Lifecycle(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + opts []mock.BrokerOption + }{ + "Embedded broker starts on ephemeral port and handles JetStream publish-subscribe": { + opts: nil, + }, + "Embedded broker starts on configured custom host": { + opts: []mock.BrokerOption{ + mock.WithBrokerHost("127.0.0.1"), + }, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + broker, err := mock.NewEmbeddedBroker(tc.opts...) + if err != nil { + t.Fatalf("NewEmbeddedBroker returned unexpected error: %v", err) + } + + if !broker.Ready() { + t.Fatal("expected broker.Ready() to be true") + } + + clientURL := broker.ClientURL() + if clientURL == "" { + t.Fatal("broker.ClientURL() is empty") + } + + nc, err := nats.Connect(clientURL) + if err != nil { + t.Fatalf("failed to connect to embedded NATS broker: %v", err) + } + + js, err := jetstream.New(nc) + if err != nil { + nc.Close() + t.Fatalf("failed to create jetstream context: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + streamName := "TEST_STREAM" + _, err = js.CreateStream(ctx, jetstream.StreamConfig{ + Name: streamName, + Subjects: []string{"test.subject"}, + }) + if err != nil { + nc.Close() + t.Fatalf("failed to create jetstream stream: %v", err) + } + + pubAck, err := js.Publish(ctx, "test.subject", []byte("payload")) + if err != nil { + nc.Close() + t.Fatalf("failed to publish to stream: %v", err) + } + if pubAck.Stream != streamName { + t.Errorf("pubAck stream mismatch: want %s, got %s", streamName, pubAck.Stream) + } + + nc.Close() + + if err := broker.Close(); err != nil { + t.Fatalf("broker.Close() returned error: %v", err) + } + + if broker.Ready() { + t.Error("expected broker.Ready() to be false after Close") + } + + // Connecting after Close must fail + _, err = nats.Connect(clientURL, nats.Timeout(200*time.Millisecond)) + if err == nil { + t.Error("expected nats.Connect to fail on closed broker, got nil error") + } + }) + } +} + +func unmarshalJSON(t *testing.T, b []byte) any { + t.Helper() + var v any + if err := json.Unmarshal(b, &v); err != nil { + t.Fatalf("failed to unmarshal json: %v", err) + } + return v +} From 94bc138af6247dbbb7c1f456ddb118b7cc4b2538 Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:29:17 -0400 Subject: [PATCH 5/6] feat(calipers): wire step, pipeline, and mock verbs into CLI --- cmd/calipers/main_test.go | 42 ++++- internal/calipers/calipers.go | 267 ++++++++++++++++++++++++++++- internal/calipers/calipers_test.go | 259 +++++++++++++++++++++++++++- 3 files changed, 563 insertions(+), 5 deletions(-) diff --git a/cmd/calipers/main_test.go b/cmd/calipers/main_test.go index 3763243..3dc2b99 100644 --- a/cmd/calipers/main_test.go +++ b/cmd/calipers/main_test.go @@ -66,7 +66,7 @@ func run(t *testing.T, dir string, args ...string) (stdout, stderr string, exitC // test drives the compiled binary as a subprocess, not the package) so it // is pinned here too; a change to calipers.go's topUsage without a matching // change here fails this test rather than passing silently. -const wantTopUsage = "usage: calipers [flags]\n" +const wantTopUsage = "usage: calipers [flags]\n" func TestMain_ReturnsUsageAndExitCodeTwoForBadInvocations(t *testing.T) { t.Parallel() @@ -120,3 +120,43 @@ func TestMain_RoutesArgsAndStreamsCorrectlyOnASuccessfulVerb(t *testing.T) { t.Error("wrong stdout", diff) } } + +func TestMain_SubprocessStepAndPipelineUsageErrors(t *testing.T) { + t.Parallel() + dir := t.TempDir() + + tests := map[string]struct { + args []string + wantCode int + wantErrContains string + }{ + "step with no subverb exits 2 with step usage": { + args: []string{"step"}, + wantCode: 2, + wantErrContains: "usage: calipers step [flags]", + }, + "pipeline with no flags exits 2 with pipeline usage": { + args: []string{"pipeline"}, + wantCode: 2, + wantErrContains: "usage: calipers pipeline --detection ", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + stdout, stderr, exitCode := run(t, dir, tc.args...) + + if diff := cmp.Diff(tc.wantCode, exitCode); diff != "" { + t.Errorf("wrong exit code (-want +got):\n%s", diff) + } + if !bytes.Contains([]byte(stderr), []byte(tc.wantErrContains)) { + t.Errorf("stderr does not contain %q, got: %s", tc.wantErrContains, stderr) + } + if diff := cmp.Diff("", stdout); diff != "" { + t.Errorf("stdout not empty (-want +got):\n%s", diff) + } + }) + } +} diff --git a/internal/calipers/calipers.go b/internal/calipers/calipers.go index c0b8f0b..fc1d026 100644 --- a/internal/calipers/calipers.go +++ b/internal/calipers/calipers.go @@ -6,8 +6,11 @@ import ( "flag" "fmt" "io" + "log/slog" "os" + "os/signal" "strings" + "syscall" "time" "github.com/ianeff/thump/api/v1/approval" @@ -16,11 +19,14 @@ import ( "github.com/ianeff/thump/internal/corpus" "github.com/ianeff/thump/internal/harvest" "github.com/ianeff/thump/internal/incident" + "github.com/ianeff/thump/internal/mock" + "github.com/ianeff/thump/internal/pipeline" "github.com/ianeff/thump/internal/probe" "github.com/ianeff/thump/internal/publish" "github.com/ianeff/thump/internal/rca" "github.com/ianeff/thump/internal/replay" "github.com/ianeff/thump/internal/scorecard" + "github.com/ianeff/thump/internal/step" "github.com/ianeff/thump/internal/tlsx" "github.com/ianeff/thump/internal/transcript" "github.com/ianeff/thump/internal/tune" @@ -89,13 +95,21 @@ func operatorProjection(ctx context.Context, natsURL, certFile, keyFile, caFile, // below by TestMain_RoutesEveryDocumentedVerbAndRefusesTheRest so the two // cannot drift the way trim's own usage string once did (it named only // "incidents" while the switch routed four verbs). -const topUsage = "usage: calipers [flags]" +const topUsage = "usage: calipers [flags]" // Main is calipers's entry point: routing to subcommand, then // either the machine (--json) or human (Lip Gloss) path over // the same Projection. // It returns the process exit code. func Main(args []string, stdout, stderr io.Writer) int { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + return MainContext(ctx, args, stdout, stderr) +} + +// MainContext executes calipers subcommands within an explicit context — +// cancels running mocks or long-running operations when the context expires. +func MainContext(ctx context.Context, args []string, stdout, stderr io.Writer) int { if len(args) == 0 { _, _ = fmt.Fprintln(stderr, topUsage) return 2 @@ -128,12 +142,263 @@ func Main(args []string, stdout, stderr io.Writer) int { return scorecard.Main(args[1:], stdout, stderr) case "validate": return validate.Main(args[1:], stdout, stderr) + case "step": + return runStep(ctx, args[1:], stdout, stderr) + case "pipeline": + return runPipeline(ctx, args[1:], stdout, stderr) + case "mock": + return runMock(ctx, args[1:], stdout, stderr) default: _, _ = fmt.Fprintln(stderr, topUsage) return 2 } } +func runStep(ctx context.Context, args []string, stdout, stderr io.Writer) int { + usage := "usage: calipers step [flags]" + if len(args) == 0 { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + + switch args[0] { + case "rattle": + return runStepRattle(ctx, args[1:], stdout, stderr) + case "clank": + return runStepClank(ctx, args[1:], stdout, stderr) + case "hiss": + return runStepHiss(ctx, args[1:], stdout, stderr) + case "thump": + return runStepThump(ctx, args[1:], stdout, stderr) + default: + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } +} + +func runStepRattle(ctx context.Context, args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("step rattle", flag.ContinueOnError) + fs.SetOutput(stderr) + watch := fs.String("watch", "config/dev/rattle/watch.yaml", "path to watch.yaml") + queryConfig := fs.String("query-config", "config/dev/rattle/query.yaml", "path to query.yaml") + queries := fs.String("queries", "", "alias for --query-config") + defaultProm := os.Getenv("PROM_URL") + if defaultProm == "" { + defaultProm = "http://localhost:9090" + } + promURL := fs.String("prom-url", defaultProm, "Prometheus base URL") + if err := fs.Parse(args); err != nil { + return 2 + } + qPath := *queryConfig + if *queries != "" { + qPath = *queries + } + if fs.NArg() > 0 && *watch == "config/dev/rattle/watch.yaml" { + *watch = fs.Arg(0) + } + if fs.NArg() > 1 && qPath == "config/dev/rattle/query.yaml" { + qPath = fs.Arg(1) + } + if fs.NArg() > 2 && *promURL == defaultProm { + *promURL = fs.Arg(2) + } + + det, err := step.RunRattle(ctx, *watch, qPath, *promURL) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + + enc := json.NewEncoder(stdout) + enc.SetIndent("", " ") + if err := enc.Encode(det); err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + return 0 +} + +func runStepClank(ctx context.Context, args []string, stdout, stderr io.Writer) int { + usage := "usage: calipers step clank --detection [--profile dir] [--model name] [--api-key key]" + fs := flag.NewFlagSet("step clank", flag.ContinueOnError) + fs.SetOutput(stderr) + detection := fs.String("detection", "", "path to detection JSON/YAML file") + profile := fs.String("profile", "config/dev", "path to profile directory") + model := fs.String("model", "haiku", "Anthropic model name") + apiKey := fs.String("api-key", os.Getenv("ANTHROPIC_API_KEY"), "Anthropic API key (defaults to ANTHROPIC_API_KEY env var)") + if err := fs.Parse(args); err != nil { + return 2 + } + if *detection == "" && fs.NArg() > 0 { + *detection = fs.Arg(0) + } + if *detection == "" { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + + ps, err := step.RunClank(ctx, *detection, *profile, *model, *apiKey) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + + enc := json.NewEncoder(stdout) + enc.SetIndent("", " ") + if err := enc.Encode(ps); err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + return 0 +} + +func runStepHiss(ctx context.Context, args []string, stdout, stderr io.Writer) int { + usage := "usage: calipers step hiss --proposal [--policy ]" + fs := flag.NewFlagSet("step hiss", flag.ContinueOnError) + fs.SetOutput(stderr) + propFile := fs.String("proposal", "", "path to proposal JSON/YAML file") + polFile := fs.String("policy", "config/dev/hiss/policy.yaml", "path to policy.yaml file") + if err := fs.Parse(args); err != nil { + return 2 + } + if *propFile == "" && fs.NArg() > 0 { + *propFile = fs.Arg(0) + } + if *polFile == "config/dev/hiss/policy.yaml" && fs.NArg() > 1 { + *polFile = fs.Arg(1) + } + if *propFile == "" { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + + gov, err := step.RunHiss(ctx, *propFile, *polFile) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + + enc := json.NewEncoder(stdout) + enc.SetIndent("", " ") + if err := enc.Encode(gov); err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + return 0 +} + +func runStepThump(ctx context.Context, args []string, stdout, stderr io.Writer) int { + usage := "usage: calipers step thump --decision [--catalog ] [--dry-run=true|false]" + fs := flag.NewFlagSet("step thump", flag.ContinueOnError) + fs.SetOutput(stderr) + decFile := fs.String("decision", "", "path to decision JSON/YAML file") + catFile := fs.String("catalog", "config/dev/actions/catalog.yaml", "path to catalog.yaml file") + dryRun := fs.Bool("dry-run", true, "dry-run execution mode") + if err := fs.Parse(args); err != nil { + return 2 + } + if *decFile == "" && fs.NArg() > 0 { + *decFile = fs.Arg(0) + } + if *catFile == "config/dev/actions/catalog.yaml" && fs.NArg() > 1 { + *catFile = fs.Arg(1) + } + if *decFile == "" { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + + out, err := step.RunThump(ctx, *decFile, *catFile, *dryRun) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + + enc := json.NewEncoder(stdout) + enc.SetIndent("", " ") + if err := enc.Encode(out); err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + return 0 +} + +func runPipeline(ctx context.Context, args []string, stdout, stderr io.Writer) int { + usage := "usage: calipers pipeline --detection [--profile dir] [--model name] [--api-key key]" + fs := flag.NewFlagSet("pipeline", flag.ContinueOnError) + fs.SetOutput(stderr) + detection := fs.String("detection", "", "path to detection JSON/YAML file") + profile := fs.String("profile", "config/dev", "path to profile directory") + model := fs.String("model", "haiku", "Anthropic model name") + apiKey := fs.String("api-key", os.Getenv("ANTHROPIC_API_KEY"), "Anthropic API key (defaults to ANTHROPIC_API_KEY env var)") + if err := fs.Parse(args); err != nil { + return 2 + } + if *detection == "" && fs.NArg() > 0 { + *detection = fs.Arg(0) + } + if *detection == "" { + _, _ = fmt.Fprintln(stderr, usage) + return 2 + } + + res, err := pipeline.Run(ctx, *detection, *profile, *model, *apiKey) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + + enc := json.NewEncoder(stdout) + enc.SetIndent("", " ") + if err := enc.Encode(res); err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + return 0 +} + +func runMock(ctx context.Context, args []string, stdout, stderr io.Writer) int { + fs := flag.NewFlagSet("mock", flag.ContinueOnError) + fs.SetOutput(stderr) + promPort := fs.Int("prom-port", 9090, "HTTP port for Prometheus/Loki telemetry stub server") + enableNATS := fs.Bool("nats", false, "start embedded NATS JetStream broker") + natsPort := fs.Int("nats-port", 4222, "TCP port for embedded NATS server") + if err := fs.Parse(args); err != nil { + return 2 + } + + ts, err := mock.NewTelemetryServer(mock.WithPort(*promPort)) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + defer func() { + if err := ts.Close(); err != nil { + slog.Warn("close mock telemetry server", "err", err) + } + }() + + _, _ = fmt.Fprintf(stdout, "mock telemetry server listening on %s\n", ts.URL()) + + if *enableNATS { + brk, err := mock.NewEmbeddedBroker(mock.WithBrokerPort(*natsPort)) + if err != nil { + _, _ = fmt.Fprintln(stderr, "calipers:", err) + return 1 + } + defer func() { + if err := brk.Close(); err != nil { + slog.Warn("close mock broker", "err", err) + } + }() + _, _ = fmt.Fprintf(stdout, "mock nats broker listening on %s\n", brk.ClientURL()) + } + + <-ctx.Done() + return 0 +} + func runIncidents(args []string, stdout, stderr io.Writer) int { var fp string rest := args diff --git a/internal/calipers/calipers_test.go b/internal/calipers/calipers_test.go index e8fa432..147b811 100644 --- a/internal/calipers/calipers_test.go +++ b/internal/calipers/calipers_test.go @@ -5,7 +5,10 @@ import ( "context" "encoding/base64" "encoding/json" + "fmt" "io" + "net/http" + "net/http/httptest" "os" "path/filepath" "strings" @@ -15,6 +18,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/ianeff/thump/api/v1/approval" "github.com/ianeff/thump/api/v1/decision" + "github.com/ianeff/thump/api/v1/outcome" "github.com/ianeff/thump/api/v1/proposal" "github.com/ianeff/thump/api/v1/signal" "github.com/ianeff/thump/internal/broker" @@ -469,7 +473,7 @@ func TestMain_ForcePublishesToThumpDecisionsOverNATSWhenNATSURLIsSet(t *testing. // and TestMain_ReturnsUsageErrorForAnUndocumentedVerbExactly below fail // immediately, which is the drift detector — not a second source of truth // to keep in sync by hand. -const wantTopUsage = "usage: calipers [flags]\n" +const wantTopUsage = "usage: calipers [flags]\n" // TestMain_RoutesEveryDocumentedVerbAndRefusesTheRest pins the usage string // and the switch together. They drifted apart in trim already — the usage @@ -486,10 +490,12 @@ func TestMain_RoutesEveryDocumentedVerbAndRefusesTheRest(t *testing.T) { t.Setenv("THUMP_SEAL_KEY", "") // corpus checks this before anything network-shaped t.Setenv("ANTHROPIC_API_KEY", "") - for _, verb := range []string{"incidents", "approve", "force", "unseal", "corpus", "rca", "tune", "replay", "harvest", "probe", "transcript", "scorecard", "validate"} { + for _, verb := range []string{"incidents", "approve", "force", "unseal", "corpus", "rca", "tune", "replay", "harvest", "probe", "transcript", "scorecard", "validate", "step", "pipeline", "mock"} { t.Run(verb, func(t *testing.T) { var out, errOut bytes.Buffer - calipers.Main([]string{verb}, &out, &errOut) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + calipers.MainContext(ctx, []string{verb}, &out, &errOut) if errOut.String() == wantTopUsage { t.Errorf("verb %q fell through to the default case — it is documented in topUsage but not routed in the switch", verb) @@ -525,3 +531,250 @@ func TestMain_ReturnsUsageErrorForAnUndocumentedVerbExactly(t *testing.T) { t.Error("wrong usage line for an undocumented verb (-want +got)", diff) } } + +func TestMain_StepRattleRunsIsolatedRattleAndPrintsJSON(t *testing.T) { + t.Parallel() + watchPath := filepath.Join("..", "..", "config", "dev", "rattle", "watch.yaml") + queryPath := filepath.Join("..", "..", "config", "dev", "rattle", "query.yaml") + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + now := time.Now().Unix() + query := r.URL.Query().Get("query") + if query == `slo:current_burn_rate:ratio{sloth_id="cart-availability"}` { + resp := fmt.Sprintf(`{ + "status": "success", + "data": { + "resultType": "matrix", + "result": [{ + "metric": {"__name__": "slo:current_burn_rate:ratio", "sloth_id": "cart-availability"}, + "values": [ + [%d, "1.0"], + [%d, "2.0"], + [%d, "4.0"], + [%d, "8.0"] + ] + }] + } + }`, now-30, now-20, now-10, now) + _, _ = w.Write([]byte(resp)) + return + } + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[]}}`)) + })) + t.Cleanup(srv.Close) + + var stdout, stderr bytes.Buffer + code := calipers.MainContext(t.Context(), []string{ + "step", "rattle", + "--watch", watchPath, + "--query-config", queryPath, + "--prom-url", srv.URL, + }, &stdout, &stderr) + + if code != 0 { + t.Fatalf("want exit code 0, got %d (stderr: %s)", code, stderr.String()) + } + + var got signal.Detection + if err := json.Unmarshal(stdout.Bytes(), &got); err != nil { + t.Fatalf("stdout is not valid JSON: %v (raw: %s)", err, stdout.String()) + } + if diff := cmp.Diff("cart", got.OriginService); diff != "" { + t.Errorf("wrong origin service (-want +got):\n%s", diff) + } + if diff := cmp.Diff("cart-availability", got.SLORef); diff != "" { + t.Errorf("wrong sloRef (-want +got):\n%s", diff) + } +} + +func TestMain_StepHissRunsIsolatedHissAndPrintsJSON(t *testing.T) { + t.Parallel() + proposalPath := filepath.Join("..", "..", "test", "fixtures", "proposals", "cart-failure.json") + policyPath := filepath.Join("..", "..", "config", "dev", "hiss", "policy.yaml") + + var stdout, stderr bytes.Buffer + code := calipers.MainContext(t.Context(), []string{ + "step", "hiss", + "--proposal", proposalPath, + "--policy", policyPath, + }, &stdout, &stderr) + + if code != 0 { + t.Fatalf("want exit code 0, got %d (stderr: %s)", code, stderr.String()) + } + + var got decision.Governed + if err := json.Unmarshal(stdout.Bytes(), &got); err != nil { + t.Fatalf("stdout is not valid JSON: %v (raw: %s)", err, stdout.String()) + } + if diff := cmp.Diff(decision.VerdictApproved, got.Decision.Verdict); diff != "" { + t.Errorf("wrong decision verdict (-want +got):\n%s", diff) + } + if diff := cmp.Diff("slo_burn:cart", got.Decision.SignalRef); diff != "" { + t.Errorf("wrong signalRef (-want +got):\n%s", diff) + } +} + +func TestMain_StepThumpRunsIsolatedThumpAndPrintsJSON(t *testing.T) { + t.Parallel() + decisionPath := filepath.Join("..", "..", "test", "fixtures", "decisions", "cart-failure.json") + catalogPath := filepath.Join("..", "..", "config", "dev", "actions", "catalog.yaml") + + var stdout, stderr bytes.Buffer + code := calipers.MainContext(t.Context(), []string{ + "step", "thump", + "--decision", decisionPath, + "--catalog", catalogPath, + "--dry-run=true", + }, &stdout, &stderr) + + if code != 0 { + t.Fatalf("want exit code 0, got %d (stderr: %s)", code, stderr.String()) + } + + var got outcome.Outcome + if err := json.Unmarshal(stdout.Bytes(), &got); err != nil { + t.Fatalf("stdout is not valid JSON: %v (raw: %s)", err, stdout.String()) + } + if diff := cmp.Diff(outcome.ModeDryRun, got.Mode); diff != "" { + t.Errorf("wrong outcome mode (-want +got):\n%s", diff) + } + if diff := cmp.Diff(outcome.ResultRendered, got.Result); diff != "" { + t.Errorf("wrong outcome result (-want +got):\n%s", diff) + } + if diff := cmp.Diff("disable-cart-failure", got.ContractRef); diff != "" { + t.Errorf("wrong contractRef (-want +got):\n%s", diff) + } +} + +func TestMain_StepUsageAndErrors(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + args []string + wantCode int + wantErrContains string + }{ + "calipers step with no subverb prints usage and returns exit code 2": { + args: []string{"step"}, + wantCode: 2, + wantErrContains: "usage: calipers step [flags]", + }, + "calipers step with unknown subverb prints usage and returns exit code 2": { + args: []string{"step", "frobnicate"}, + wantCode: 2, + wantErrContains: "usage: calipers step [flags]", + }, + "calipers step hiss with missing proposal prints usage and returns exit code 2": { + args: []string{"step", "hiss"}, + wantCode: 2, + wantErrContains: "usage: calipers step hiss", + }, + "calipers step thump with missing decision prints usage and returns exit code 2": { + args: []string{"step", "thump"}, + wantCode: 2, + wantErrContains: "usage: calipers step thump", + }, + "calipers step clank with missing detection prints usage and returns exit code 2": { + args: []string{"step", "clank"}, + wantCode: 2, + wantErrContains: "usage: calipers step clank", + }, + "calipers step hiss with non-existent proposal returns exit code 1": { + args: []string{"step", "hiss", "--proposal", "nonexistent.json"}, + wantCode: 1, + wantErrContains: "calipers:", + }, + "calipers step thump with non-existent decision returns exit code 1": { + args: []string{"step", "thump", "--decision", "nonexistent.json"}, + wantCode: 1, + wantErrContains: "calipers:", + }, + "calipers step clank with nonexistent detection file returns exit code 1": { + args: []string{"step", "clank", "--detection", "nonexistent.json", "--profile", filepath.Join("..", "..", "config", "dev")}, + wantCode: 1, + wantErrContains: "calipers:", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + var stdout, stderr bytes.Buffer + code := calipers.MainContext(t.Context(), tc.args, &stdout, &stderr) + + if diff := cmp.Diff(tc.wantCode, code); diff != "" { + t.Errorf("wrong exit code (-want +got):\n%s", diff) + } + if !strings.Contains(stderr.String(), tc.wantErrContains) { + t.Errorf("stderr does not contain %q, got: %s", tc.wantErrContains, stderr.String()) + } + }) + } +} + +func TestMain_PipelineUsageAndErrors(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + args []string + wantCode int + wantErrContains string + }{ + "calipers pipeline with no args prints usage and returns exit code 2": { + args: []string{"pipeline"}, + wantCode: 2, + wantErrContains: "usage: calipers pipeline --detection ", + }, + "calipers pipeline with nonexistent detection file returns exit code 1": { + args: []string{"pipeline", "--detection", "nonexistent.json"}, + wantCode: 1, + wantErrContains: "calipers:", + }, + "calipers pipeline with nonexistent profile directory returns exit code 1": { + args: []string{"pipeline", "--detection", filepath.Join("..", "..", "test", "fixtures", "detections", "cart-failure.json"), "--profile", "nonexistent-dir"}, + wantCode: 1, + wantErrContains: "calipers:", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + var stdout, stderr bytes.Buffer + code := calipers.MainContext(t.Context(), tc.args, &stdout, &stderr) + + if diff := cmp.Diff(tc.wantCode, code); diff != "" { + t.Errorf("wrong exit code (-want +got):\n%s", diff) + } + if !strings.Contains(stderr.String(), tc.wantErrContains) { + t.Errorf("stderr does not contain %q, got: %s", tc.wantErrContains, stderr.String()) + } + }) + } +} + +func TestMain_MockStartsTelemetryAndBroker(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + time.AfterFunc(200*time.Millisecond, cancel) + + var stdout, stderr bytes.Buffer + code := calipers.MainContext(ctx, []string{ + "mock", + "--prom-port", "0", + "--nats", + "--nats-port", "-1", + }, &stdout, &stderr) + + if code != 0 { + t.Fatalf("want exit code 0, got %d (stderr: %s)", code, stderr.String()) + } + out := stdout.String() + if !strings.Contains(out, "mock telemetry server listening on") { + t.Errorf("stdout does not mention telemetry server, got: %s", out) + } + if !strings.Contains(out, "mock nats broker listening on") { + t.Errorf("stdout does not mention nats broker, got: %s", out) + } +} From 378ec0fc8398430eb780a8638c4559e73d9b17f0 Mon Sep 17 00:00:00 2001 From: Ian Furst Date: Wed, 19 Aug 2026 14:34:45 -0400 Subject: [PATCH 6/6] feat(taskfile,docs): add beat and dev targets for isolated testing and update docs --- Taskfile.yaml | 57 ++++++++++++++++++++++++++++++++ cmd/calipers/main_test.go | 3 +- docs/dev-environment.md | 69 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 1 deletion(-) diff --git a/Taskfile.yaml b/Taskfile.yaml index 338687e..40026ad 100644 --- a/Taskfile.yaml +++ b/Taskfile.yaml @@ -422,6 +422,44 @@ tasks: cmds: - go run ./cmd/calipers + beat:rattle: + desc: "Run rattle in isolated single-beat mode against Prometheus: task beat:rattle [WATCH=config/dev/rattle/watch.yaml] [QUERY=config/dev/rattle/query.yaml] [PROM_URL=http://localhost:9090]" + vars: + WATCH: '{{.WATCH | default "config/dev/rattle/watch.yaml"}}' + QUERY: '{{.QUERY | default "config/dev/rattle/query.yaml"}}' + PROM_URL: '{{.PROM_URL | default "http://localhost:9090"}}' + cmds: + - go run ./cmd/calipers step rattle --watch {{.WATCH}} --query-config {{.QUERY}} --prom-url {{.PROM_URL}} + + beat:clank: + desc: "Run clank in isolated single-beat mode against a detection: task beat:clank DETECTION= [PROFILE=config/dev] [MODEL=haiku] (key-gated)" + requires: + vars: [DETECTION] + vars: + PROFILE: '{{.PROFILE | default "config/dev"}}' + MODEL: '{{.MODEL | default "haiku"}}' + cmds: + - go run ./cmd/calipers step clank --detection {{.DETECTION}} --profile {{.PROFILE}} --model {{.MODEL}} + + beat:hiss: + desc: "Run hiss in isolated single-beat mode against a proposal: task beat:hiss PROPOSAL= [POLICY=config/dev/hiss/policy.yaml]" + requires: + vars: [PROPOSAL] + vars: + POLICY: '{{.POLICY | default "config/dev/hiss/policy.yaml"}}' + cmds: + - go run ./cmd/calipers step hiss --proposal {{.PROPOSAL}} --policy {{.POLICY}} + + beat:thump: + desc: "Run thump in isolated single-beat mode against a decision: task beat:thump DECISION= [CATALOG=config/dev/actions/catalog.yaml] [DRY_RUN=true]" + requires: + vars: [DECISION] + vars: + CATALOG: '{{.CATALOG | default "config/dev/actions/catalog.yaml"}}' + DRY_RUN: '{{.DRY_RUN | default "true"}}' + cmds: + - go run ./cmd/calipers step thump --decision {{.DECISION}} --catalog {{.CATALOG}} --dry-run={{.DRY_RUN}} + # docs/dev-environment.md — the fully-local dev environment. dev:cluster is # the one piece Tilt itself can't do: `allow_k8s_contexts()` and `helm()` # both evaluate at Tiltfile load, so the k3d context has to exist before @@ -564,6 +602,25 @@ tasks: - kubectl --context k3d-thump-dev -n thump delete job dev-govern-{{.RUN_ID}} configmap dev-govern-input-{{.RUN_ID}} --ignore-not-found - rm -rf bin/govern/{{.RUN_ID}} + dev:pipeline: + desc: "Run headless end-to-end pipeline locally without cluster: task dev:pipeline DETECTION= [PROFILE=config/dev] [MODEL=haiku] (key-gated)" + requires: + vars: [DETECTION] + vars: + PROFILE: '{{.PROFILE | default "config/dev"}}' + MODEL: '{{.MODEL | default "haiku"}}' + cmds: + - go run ./cmd/calipers pipeline --detection {{.DETECTION}} --profile {{.PROFILE}} --model {{.MODEL}} + + dev:mock: + desc: "Run lightweight telemetry and NATS mock server: task dev:mock [PROM_PORT=9090] [NATS=false] [NATS_PORT=4222]" + vars: + PROM_PORT: '{{.PROM_PORT | default "9090"}}' + NATS_PORT: '{{.NATS_PORT | default "4222"}}' + NATS: '{{.NATS | default "false"}}' + cmds: + - go run ./cmd/calipers mock --prom-port {{.PROM_PORT}} {{if eq .NATS "true"}}--nats --nats-port {{.NATS_PORT}}{{end}} + dev:probe: desc: >- Fire N real clank reasoning runs against a captured detection, straight diff --git a/cmd/calipers/main_test.go b/cmd/calipers/main_test.go index 3dc2b99..e32a067 100644 --- a/cmd/calipers/main_test.go +++ b/cmd/calipers/main_test.go @@ -6,6 +6,7 @@ import ( "os" "os/exec" "path/filepath" + "strings" "testing" "github.com/google/go-cmp/cmp" @@ -151,7 +152,7 @@ func TestMain_SubprocessStepAndPipelineUsageErrors(t *testing.T) { if diff := cmp.Diff(tc.wantCode, exitCode); diff != "" { t.Errorf("wrong exit code (-want +got):\n%s", diff) } - if !bytes.Contains([]byte(stderr), []byte(tc.wantErrContains)) { + if !strings.Contains(stderr, tc.wantErrContains) { t.Errorf("stderr does not contain %q, got: %s", tc.wantErrContains, stderr) } if diff := cmp.Diff("", stdout); diff != "" { diff --git a/docs/dev-environment.md b/docs/dev-environment.md index ccddf3c..babc186 100644 --- a/docs/dev-environment.md +++ b/docs/dev-environment.md @@ -18,6 +18,75 @@ a separate repo under `~/projects/ceph/`; `dev` is provisioned by this one, via you. A five-beat cycle on Haiku costs fractions of a cent; nothing about running this environment burns meaningful spend. +## Isolated beat testing & lightweight development (zero cluster / low resource) + +When iterating on a single beat, testing policy logic, or running reasoning without provisioning a k3d cluster or Docker VM, `calipers` provides single-beat execution steps, a headless end-to-end pipeline runner, and a lightweight mock server. + +### Single-beat execution + +Drive individual beats directly from fixtures without NATS or Kubernetes: + +| Beat | Task / Command | Input | Output | +|---|---|---|---| +| rattle | `task beat:rattle [PROM_URL=http://localhost:9090]` | PromQL queries against Prometheus | `signal.Detection` JSON | +| clank | `task beat:clank DETECTION= [MODEL=haiku]` | Detection fixture + profile configs | `proposal.Set` JSON | +| hiss | `task beat:hiss PROPOSAL= [POLICY=]` | Proposal fixture + policy YAML | `decision.Governed` JSON | +| thump | `task beat:thump DECISION= [CATALOG=] [DRY_RUN=true]` | Decision fixture + catalog YAML | `outcome.Outcome` JSON | + +Equivalent direct CLI invocations: + +```sh +# rattle: detect SLO burn against Prometheus (or dev:mock) +go run ./cmd/calipers step rattle --watch config/dev/rattle/watch.yaml --query-config config/dev/rattle/query.yaml --prom-url http://localhost:9090 + +# clank: reason over a captured detection (requires ANTHROPIC_API_KEY) +go run ./cmd/calipers step clank --detection test/fixtures/detections/cart-failure.json --profile config/dev --model haiku + +# hiss: evaluate governance policy against a proposal +go run ./cmd/calipers step hiss --proposal test/fixtures/proposals/cart-failure.json --policy config/dev/hiss/policy.yaml + +# thump: dry-run action synthesis and safety bounds against a decision +go run ./cmd/calipers step thump --decision test/fixtures/decisions/cart-failure.json --catalog config/dev/actions/catalog.yaml --dry-run=true +``` + +### Headless pipeline execution + +Run the full five-beat loop in-process against local configuration profiles and live reasoning: + +```sh +task dev:pipeline DETECTION=test/fixtures/detections/cart-failure.json +``` + +Or via CLI: + +```sh +go run ./cmd/calipers pipeline --detection test/fixtures/detections/cart-failure.json --profile config/dev --model haiku +``` + +This runs detection ingestion, clank reasoning, hiss policy governance, and thump dry-run actuation in a single process without NATS or Kubernetes, printing the complete `pipeline.Result` JSON to stdout. + +### Lightweight mock telemetry & broker + +For local testing without the 12 GB substrate, `dev:mock` stands up an in-process HTTP stub for Prometheus/Loki and an optional embedded NATS JetStream server: + +```sh +task dev:mock PROM_PORT=9090 +``` + +With embedded NATS: + +```sh +task dev:mock PROM_PORT=9090 NATS=true NATS_PORT=4222 +``` + +Direct CLI invocation: + +```sh +go run ./cmd/calipers mock --prom-port 9090 --nats --nats-port 4222 +``` + +The mock telemetry server returns deterministic PromQL vector and matrix responses (such as `slo:current_burn_rate:ratio`), fake Loki query streams, and Kubernetes pod list stubs for telemetry tools in clank and rattle. + ## Bringing it up ```sh