fix: Step Functions contract — SLA cancel, run IDs, deadlines, ASL guard

CHANGELOG.md

@@ -34,6 +34,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Stream router correlation IDs** — Per-record correlation IDs injected into context for structured log tracing across services.
 - **Telemetry flush per invocation** — OTel providers flush (not shutdown) per Lambda invocation to survive environment reuse across warm starts.
 
+### Fixed
+
+- **SFN SLA cancel states no longer fail with `States.Runtime`** — `CancelSLASchedules` and `CancelSLAOnCompleteTriggerFailure` referenced `$.config.sla.deadline`, `$.config.sla.expectedDuration`, `$.config.sla.maxDuration` and `$.sensorArrivalAt`, but `types.SLAConfig` marks every field `omitempty`, so an absolute-only SLA never emitted `maxDuration`, a relative-only SLA never emitted `deadline`, and `sensorArrivalAt` was usually absent. The resulting `States.Runtime` is not retriable and is not caught by `Catch: ["States.ALL"]`, so every SLA-configured execution failed after `CompleteTrigger`. The SFN input now uses dedicated `SFNInput`/`SFNConfig`/`SFNSLA` types that always emit those keys, and both cancel states now forward `timezone` so `handleSLACancel` recomputes deadlines in the configured zone.
+- **Trigger run IDs are extracted for every trigger type** — `ExtractRunID` only matched `runId`, `jobRunId`, `glue_job_run_id`, `executionArn`, `stepId` and `dagRunId`, while the step-function, EMR, EMR Serverless, Airflow and Databricks executors emit `sfn_execution_arn`, `emr_step_id`, `emr_sl_job_run_id`, `airflow_dag_run_id` and `databricks_run_id`. The empty run ID was then omitted from the Lambda result and the `CheckJob` state raised `States.Runtime` after the external job had already been launched. `OrchestratorOutput.RunID` is now always marshaled and an empty metadata map takes the sync-sentinel path.
+- **Absolute SLA deadlines are anchored to the execution date** — `CalculateAbsoluteDeadline` rolled an explicit execution date forward by 24h (or 1h for `:MM` deadlines) whenever the deadline had already passed. `sla-monitor` `cancel` therefore published `SLA_MET` for runs that finished late, the `reconcile` breach branch was unreachable, and the watchdog scheduled breach alerts a day late. Roll-forward now applies only when no execution date is supplied, and uses `AddDate` so the wall-clock time survives DST transitions.
+- **Orchestrator evaluate/trigger results satisfy the state machine contract** — `handleEvaluate` returned a result without `status` on storage failures, so the `IsReady` Choice dereferenced a missing `$.evaluateResult.status`; it now always emits a status. `handleTrigger` returned a partial result with a nil error on configuration failures, so `HasTriggerResult` saw `IsPresent=true` and `CheckJob` dereferenced a missing `runId`, killing the execution with the `TRIGGER#` lock stuck in `RUNNING` until TTL; those paths now return a Lambda error so `Trigger`'s Retry/Catch routes to `TriggerRetryExhausted`, which releases the lock.
+- **Time-dependent tests** — `TestSLAMonitor_Calculate_ReturnsRFC3339` (failing since 2026-06-16), `TestSLAMonitor_Reconcile_ReturnsDeadlines`, `TestSLAMonitor_Cancel_RecalculatesWhenTimesNotProvided` and three watchdog proactive-SLA tests now inject `Deps.NowFunc` instead of relying on the wall clock.
+- **SLA cancel verdict uses the T+1 date for sensor-triggered daily pipelines** — `handleSLACancel` and the dry-run SLA projection recomputed the absolute deadline from `input.Date`/`date` (the data date D), but sensor-triggered daily pipelines run T+1 — data for date D arrives on D+1, and the watchdog's proactive SLA scheduling already anchors the deadline to D+1. Now that an explicit date no longer rolls forward when its deadline has passed, `cancel` published a false `SLA_BREACH` for pipelines that finished on D+1 before their deadline. The new `internal/lambda.ResolveSLADate` centralizes the watchdog's T+1 rule (cron pipelines and hourly `:MM` deadlines are unaffected) and is now shared by the watchdog, `sla.handleSLACancel`, and `stream.publishDryRunSLAProjection`. `orchestrator.handleEvaluate` storage failures (`GetConfig` error, config not found, `GetAllSensors` error) are now also logged at error level instead of surfacing only in the returned `status: "error"` payload.
+
+**Release notes**: verdicts for pipelines with a non-UTC `sla.timezone` are now computed in that zone; absolute SLA deadlines no longer roll forward to the next day when past; in-flight executions started before deploy are unaffected by the state-machine change; the sla-monitor `reconcile` mode has no production invoker today and is not a safety net for deadlines missed while the watchdog was down.
+
 ### Dependencies
 
 - `go.opentelemetry.io/otel` v1.43.0 (traces + metrics)

deploy/statemachine.asl.json

@@ -397,6 +397,7 @@
         "deadline.$": "$.config.sla.deadline",
         "expectedDuration.$": "$.config.sla.expectedDuration",
         "maxDuration.$": "$.config.sla.maxDuration",
+        "timezone.$": "$.config.sla.timezone",
         "sensorArrivalAt.$": "$.sensorArrivalAt"
       },
       "ResultPath": "$.slaResult",
@@ -451,6 +452,7 @@
         "deadline.$": "$.config.sla.deadline",
         "expectedDuration.$": "$.config.sla.expectedDuration",
         "maxDuration.$": "$.config.sla.maxDuration",
+        "timezone.$": "$.config.sla.timezone",
         "sensorArrivalAt.$": "$.sensorArrivalAt"
       },
       "ResultPath": "$.slaResult",

deploy/statemachine_contract_test.go

@@ -0,0 +1,362 @@
+package deploy_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dwsmith1983/interlock/internal/lambda"
+	"github.com/dwsmith1983/interlock/internal/lambda/orchestrator"
+	"github.com/dwsmith1983/interlock/internal/store/storetest"
+	"github.com/dwsmith1983/interlock/pkg/types"
+)
+
+// --- JSONPath resolution ---------------------------------------------------
+
+// pathRefRE matches an ASL reference path such as $.a.b[0].c. Used to pull the
+// arguments out of intrinsic calls like States.MathAdd($.a, $.b).
+var pathRefRE = regexp.MustCompile(`\$(?:\.[A-Za-z0-9_]+(?:\[\d+\])*)+`)
+
+// resolvePath walks a reference path against a decoded JSON document and
+// reports whether it exists. Supports "$", "$.a.b" and "$.a[0].b".
+func resolvePath(doc interface{}, path string) bool {
+	if path == "$" {
+		return true
+	}
+	if !strings.HasPrefix(path, "$.") {
+		return false
+	}
+	cur := doc
+	for _, seg := range strings.Split(strings.TrimPrefix(path, "$."), ".") {
+		name, indices := splitSegment(seg)
+		if name != "" {
+			m, ok := cur.(map[string]interface{})
+			if !ok {
+				return false
+			}
+			v, ok := m[name]
+			if !ok {
+				return false
+			}
+			cur = v
+		}
+		for _, idx := range indices {
+			arr, ok := cur.([]interface{})
+			if !ok || idx < 0 || idx >= len(arr) {
+				return false
+			}
+			cur = arr[idx]
+		}
+	}
+	return true
+}
+
+// splitSegment splits "items[0][1]" into ("items", []int{0, 1}).
+func splitSegment(seg string) (name string, indices []int) {
+	name = seg
+	if i := strings.Index(seg, "["); i >= 0 {
+		name = seg[:i]
+		for _, raw := range strings.Split(strings.Trim(seg[i:], "[]"), "][") {
+			n, err := strconv.Atoi(raw)
+			if err != nil {
+				continue
+			}
+			indices = append(indices, n)
+		}
+	}
+	return name, indices
+}
+
+// --- ASL reference collection ----------------------------------------------
+
+// aslState is the subset of a state definition that carries reference paths.
+type aslState struct {
+	Type        string                   `json:"Type"`
+	Parameters  map[string]interface{}   `json:"Parameters"`
+	SecondsPath string                   `json:"SecondsPath"`
+	Choices     []map[string]interface{} `json:"Choices"`
+}
+
+// requiredPaths returns every reference path the state dereferences
+// unconditionally. A missing path raises States.Runtime at runtime.
+func requiredPaths(st aslState) []string {
+	out := collectParameterPaths(st.Parameters)
+	if strings.HasPrefix(st.SecondsPath, "$") {
+		out = append(out, st.SecondsPath)
+	}
+	for _, rule := range st.Choices {
+		out = append(out, requiredChoicePaths(rule)...)
+	}
+	return out
+}
+
+// collectParameterPaths walks a Parameters object and returns every reference
+// used by a "<key>.$" entry, including paths inside intrinsic calls.
+func collectParameterPaths(params map[string]interface{}) []string {
+	var out []string
+	for k, v := range params {
+		if !strings.HasSuffix(k, ".$") {
+			if nested, ok := v.(map[string]interface{}); ok {
+				out = append(out, collectParameterPaths(nested)...)
+			}
+			continue
+		}
+		s, ok := v.(string)
+		if !ok {
+			continue
+		}
+		if strings.HasPrefix(s, "States.") {
+			out = append(out, pathRefRE.FindAllString(s, -1)...)
+			continue
+		}
+		if strings.HasPrefix(s, "$") {
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+// requiredChoicePaths returns the reference paths a Choice rule dereferences
+// unconditionally. A Variable in a rule that also carries IsPresent is
+// explicitly allowed to be absent — that is what IsPresent is for.
+func requiredChoicePaths(rule map[string]interface{}) []string {
+	if _, guarded := rule["IsPresent"]; guarded {
+		return nil
+	}
+	var out []string
+	for k, v := range rule {
+		switch k {
+		case "Variable":
+			if s, ok := v.(string); ok && strings.HasPrefix(s, "$") {
+				out = append(out, s)
+			}
+		case "Not":
+			if m, ok := v.(map[string]interface{}); ok {
+				out = append(out, requiredChoicePaths(m)...)
+			}
+		case "And", "Or":
+			if arr, ok := v.([]interface{}); ok {
+				for _, e := range arr {
+					if m, ok := e.(map[string]interface{}); ok {
+						out = append(out, requiredChoicePaths(m)...)
+					}
+				}
+			}
+		default:
+			// Comparator paths such as NumericGreaterThanEqualsPath.
+			if strings.HasSuffix(k, "Path") {
+				if s, ok := v.(string); ok && strings.HasPrefix(s, "$") {
+					out = append(out, s)
+				}
+			}
+		}
+	}
+	return out
+}
+
+// --- real handler outputs ---------------------------------------------------
+
+// contractConfigItem builds the control-table row that store.GetConfig
+// expects. Every test in this file uses pipeline "gold-orders".
+func contractConfigItem(t *testing.T, cfg types.PipelineConfig) map[string]ddbtypes.AttributeValue {
+	t.Helper()
+	data, err := json.Marshal(cfg)
+	require.NoError(t, err)
+	return map[string]ddbtypes.AttributeValue{
+		"PK":     &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("gold-orders")},
+		"SK":     &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK},
+		"config": &ddbtypes.AttributeValueMemberS{Value: string(data)},
+	}
+}
+
+type contractExecutor struct{ meta map[string]interface{} }
+
+func (c *contractExecutor) Execute(context.Context, *types.TriggerConfig) (map[string]interface{}, error) {
+	return c.meta, nil
+}
+
+// evaluateErrorResult returns the output the live evaluate handler produces
+// when the control table is unavailable — the worst case for IsReady.
+func evaluateErrorResult(t *testing.T) lambda.OrchestratorOutput {
+	t.Helper()
+	fake := &storetest.FakeDynamo{
+		GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) {
+			return nil, errors.New("dynamodb: internal error")
+		},
+	}
+	d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()}
+	out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{
+		Mode: "evaluate", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01",
+	})
+	require.NoError(t, err, "evaluate must not return a Lambda error for a storage failure")
+	return out
+}
+
+// triggerResult returns the output the live trigger handler produces for a
+// polling (Glue) trigger — the state CheckJob reads.
+func triggerResult(t *testing.T) lambda.OrchestratorOutput {
+	t.Helper()
+	cfg := types.PipelineConfig{
+		Pipeline: types.PipelineIdentity{ID: "gold-orders"},
+		Job:      types.JobConfig{Type: types.TriggerGlue, Config: map[string]interface{}{"jobName": "etl"}},
+	}
+	fake := &storetest.FakeDynamo{
+		GetItemFn: func(context.Context, *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) {
+			return &dynamodb.GetItemOutput{Item: contractConfigItem(t, cfg)}, nil
+		},
+	}
+	d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()}
+	d.TriggerRunner = &contractExecutor{meta: map[string]interface{}{
+		"glue_job_name": "etl", "glue_job_run_id": "jr_1",
+	}}
+	out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{
+		Mode: "trigger", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01",
+	})
+	require.NoError(t, err)
+	return out
+}
+
+// checkJobTerminalResult returns the terminal check-job output. CompleteTrigger
+// is only reachable once IsJobDone has seen a terminal event, so this is the
+// document CompleteTrigger's Parameters are resolved against.
+func checkJobTerminalResult(t *testing.T) lambda.OrchestratorOutput {
+	t.Helper()
+	fake := &storetest.FakeDynamo{
+		QueryFn: func(context.Context, *dynamodb.QueryInput) (*dynamodb.QueryOutput, error) {
+			return &dynamodb.QueryOutput{Items: []map[string]ddbtypes.AttributeValue{{
+				"PK":    &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("gold-orders")},
+				"SK":    &ddbtypes.AttributeValueMemberS{Value: types.JobSK("daily", "2026-03-01", "1709280000000")},
+				"event": &ddbtypes.AttributeValueMemberS{Value: types.JobEventSuccess},
+			}}}, nil
+		},
+	}
+	d := &lambda.Deps{Store: storetest.NewStore(fake), Logger: slog.Default()}
+	out, err := orchestrator.HandleOrchestrator(context.Background(), d, lambda.OrchestratorInput{
+		Mode: "check-job", PipelineID: "gold-orders", ScheduleID: "daily", Date: "2026-03-01",
+	})
+	require.NoError(t, err)
+	return out
+}
+
+// stateDocument builds the JSON document Step Functions holds when each state
+// runs: the execution input plus every ResultPath written earlier on the path
+// that reaches the state. evalLoop/jobPollLoop mirror the Result blocks of the
+// InitEvalLoop and InitJobPollLoop Pass states; errorInfo mirrors the object
+// Step Functions writes for a Catch.
+func stateDocument(t *testing.T, input lambda.SFNInput) map[string]interface{} {
+	t.Helper()
+	data, err := json.Marshal(input)
+	require.NoError(t, err)
+
+	var doc map[string]interface{}
+	require.NoError(t, json.Unmarshal(data, &doc))
+
+	doc["evalLoop"] = map[string]interface{}{"elapsedSeconds": float64(0)}
+	doc["jobPollLoop"] = map[string]interface{}{"elapsedSeconds": float64(0)}
+	doc["evaluateResult"] = toDoc(t, evaluateErrorResult(t))
+	doc["triggerResult"] = toDoc(t, triggerResult(t))
+	doc["checkJobResult"] = toDoc(t, checkJobTerminalResult(t))
+	doc["errorInfo"] = map[string]interface{}{
+		"Error": "States.TaskFailed",
+		"Cause": "trigger execute: glue trigger: StartJobRun failed",
+	}
+	return doc
+}
+
+func toDoc(t *testing.T, v interface{}) map[string]interface{} {
+	t.Helper()
+	data, err := json.Marshal(v)
+	require.NoError(t, err)
+	var m map[string]interface{}
+	require.NoError(t, json.Unmarshal(data, &m))
+	return m
+}
+
+// --- the contract -----------------------------------------------------------
+
+// slaOnlyStates are reachable only when the CheckCancelSLA /
+// CheckSLAForCompleteTriggerFailure IsPresent guard on $.config.sla passes.
+var slaOnlyStates = map[string]bool{
+	"CancelSLASchedules":                true,
+	"CancelSLAOnCompleteTriggerFailure": true,
+}
+
+func contractPipelineConfig(sla *types.SLAConfig) *types.PipelineConfig {
+	return &types.PipelineConfig{
+		Pipeline: types.PipelineIdentity{ID: "gold-orders"},
+		Schedule: types.ScheduleConfig{Evaluation: types.EvaluationWindow{Window: "1h", Interval: "5m"}},
+		SLA:      sla,
+		Job:      types.JobConfig{Type: types.TriggerGlue, Config: map[string]interface{}{"jobName": "etl"}},
+	}
+}
+
+// TestASL_EveryDereferencedPathResolves is the Go<->ASL contract guard. Every
+// Parameters reference, Wait SecondsPath, Choice comparator path and
+// unguarded Choice Variable in the rendered state machine must resolve against
+// the real marshaled Go payloads. A path that does not resolve raises
+// States.Runtime at runtime, which Retry cannot retry and Catch: States.ALL
+// cannot intercept.
+func TestASL_EveryDereferencedPathResolves(t *testing.T) {
+	asl := loadASL(t)
+
+	scenarios := []struct {
+		name            string
+		sla             *types.SLAConfig
+		sensorArrivalAt string
+	}{
+		{
+			name: "absolute SLA only",
+			sla:  &types.SLAConfig{Deadline: "08:00", ExpectedDuration: "30m"},
+		},
+		{
+			name:            "relative SLA with sensor arrival",
+			sla:             &types.SLAConfig{MaxDuration: "2h"},
+			sensorArrivalAt: "2026-03-01T06:00:00Z",
+		},
+		{
+			name: "relative SLA without sensor arrival",
+			sla:  &types.SLAConfig{MaxDuration: "2h"},
+		},
+		{
+			name: "no SLA",
+			sla:  nil,
+		},
+	}
+
+	for _, sc := range scenarios {
+		t.Run(sc.name, func(t *testing.T) {
+			input := lambda.BuildSFNInput(
+				contractPipelineConfig(sc.sla), "gold-orders", "daily", "2026-03-01", sc.sensorArrivalAt)
+			doc := stateDocument(t, input)
+
+			// The SLA branch is selected by IsPresent on $.config.sla.
+			slaPresent := resolvePath(doc, "$.config.sla")
+			require.Equal(t, sc.sla != nil, slaPresent,
+				"$.config.sla presence must match whether the pipeline has an SLA")
+
+			for name, raw := range asl.States {
+				if sc.sla == nil && slaOnlyStates[name] {
+					continue // unreachable without an SLA
+				}
+				var st aslState
+				require.NoError(t, json.Unmarshal(raw, &st), "parsing state %q", name)
+
+				for _, path := range requiredPaths(st) {
+					ok := resolvePath(doc, path)
+					require.Truef(t, ok,
+						"state %q dereferences %q, which is absent from the Step Functions state document; "+
+							"this raises States.Runtime and cannot be caught", name, path)
+				}
+			}
+		})
+	}
+}