// Package workflow_test contains formal-model-derived conformance tests for
// specs/otel-observability-spec.md, Sections 13-16 (Outcome Evaluation, Local
// Mirrors and Artifacts, Security and Privacy, Reliability and Failure
// Handling).
//
// Formal predicates encoded by this file (see accompanying issue for the full
// notation):
//
// P1_OutcomeSeparateTrace - §13.1 outcome evaluator MUST NOT extend the
// original workflow trace across long spans.
// P2_OutcomeSourceCorrelation - §13.2 evaluation span carries source
// correlation attributes when a full span
// context is unavailable.
// P3_OutcomeResultTaxonomy - §13.3 gh-aw.outcome.result uses the
// canonical outcome taxonomy.
// P4_MirrorPathStable - §14.1 local mirror path is
// /tmp/gh-aw/otel.jsonl.
// P5_MirrorWriteBeforeExport - §14.3 mirror writes occur before remote
// export success is assumed.
// P6_MirrorNotTruncatedOnFail - §14.3 remote export failure MUST NOT
// delete/truncate previously written mirror
// data.
// P7_HeaderRedactionInMirror - §15.1 exporter headers/credentials MUST NOT
// appear in mirrored/artifact records.
// P8_ContentDefaultNone - §15.2 raw prompts/responses/tool args/tool
// results MUST NOT be captured by default.
// P9_MetricDimensionBound - §15.4 / §13.3 user/run/item identifiers
// MUST NOT be metric dimensions.
// P10_NonFatalExportFailure - §16.1 telemetry export failure SHOULD NOT
// change a successful functional workflow
// result into a failure.
// P11_PartialFanOutIndependent - §16.3 for N endpoints, each is attempted
// independently; partial success is not
// discarded.
// INV1_RetryBounded - §16.2 retry MUST be bounded by attempts and
// elapsed time; permanent failures MUST NOT
// retry indefinitely.
// SAFETY_FailClosedSecrets - the "Safeguards" block: export/fan-out
// failures and shutdown timeouts MUST be
// recorded as bounded diagnostics and MUST
// NOT be reported as successful delivery, nor
// expose credentials or delete mirror data.
package workflow_test
import (
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// ---------------------------------------------------------------------------
// stub — replace with real implementation
//
// No concrete Go implementation of outcome-evaluation span emission, mirror
// writing, or fan-out retry orchestration exists yet in pkg/workflow. These
// stub types model the behavior described in Sections 13-16 of
// specs/otel-observability-spec.md so the predicates can be tested today and
// swapped for real call sites once implemented.
// ---------------------------------------------------------------------------
// outcomeResult is the canonical outcome taxonomy from
// specs/safe-output-outcome-evaluation.md referenced in §13.3.
type outcomeResult string
const (
outcomeAccepted outcomeResult = "accepted"
outcomeRejected outcomeResult = "rejected"
outcomeIgnored outcomeResult = "ignored"
outcomePending outcomeResult = "pending"
outcomeLifecycle outcomeResult = "lifecycle"
outcomeLifecycleClose outcomeResult = "lifecycle_close"
)
func validOutcomeTaxonomy() map[outcomeResult]bool {
return map[outcomeResult]bool{
outcomeAccepted: true,
outcomeRejected: true,
outcomeIgnored: true,
outcomePending: true,
outcomeLifecycle: true,
outcomeLifecycleClose: true,
}
}
// outcomeSpan models an evaluation span per §13.3.
type outcomeSpan struct {
name string
kind string
result outcomeResult
sourceRunID string
sourceWorkflow string
sourceRepo string
hasSourceLink bool // full span context available -> link, not attrs
metricDimensions []string
}
// P2_OutcomeSourceCorrelation: when a full span context is unavailable, the
// evaluation span SHOULD include source_run_id/source_workflow/repo.
func (s outcomeSpan) hasSourceCorrelation() bool {
if s.hasSourceLink {
return true
}
return s.sourceRunID != "" && s.sourceWorkflow != "" && s.sourceRepo != ""
}
// P9_MetricDimensionBound: URLs and item identifiers MUST NOT be metric
// dimensions (§13.3, §15.4).
func (s outcomeSpan) violatesMetricDimensionBound() bool {
for _, d := range s.metricDimensions {
if strings.Contains(d, "://") || strings.Contains(d, "item_id") || strings.Contains(d, "run_id") {
return true
}
}
return false
}
// mirrorWriter models the local telemetry mirror writer per §14.
type mirrorWriter struct {
written []string
}
func (m *mirrorWriter) writeBeforeExport(record string) {
// P5_MirrorWriteBeforeExport: write happens unconditionally, prior to any
// export-success gating.
m.written = append(m.written, record)
}
// exportOutcome models the result of attempting export after a mirror write.
type exportOutcome int
const (
exportSuccess exportOutcome = iota
exportFailure
)
func (m *mirrorWriter) simulateExport(_ exportOutcome) []string {
// P6_MirrorNotTruncatedOnFail: regardless of export outcome, previously
// written mirror records are preserved (never deleted/truncated).
return m.written
}
// redactHeaders models §15.1: header/credential material MUST NOT appear in
// mirrored records, artifacts, diagnostics, or job summaries.
func redactHeaders(headers map[string]string) map[string]string {
redacted := make(map[string]string, len(headers))
for k := range headers {
redacted[k] = "***"
}
return redacted
}
// captureContentMode models §5.6 / §15.2 capture-content default behavior.
type captureContentMode string
const (
captureNone captureContentMode = "none"
captureMetadata captureContentMode = "metadata"
captureFull captureContentMode = "full"
)
// defaultCaptureMode returns the mode when unset, per spec default.
func defaultCaptureMode(configured captureContentMode) captureContentMode {
if configured == "" {
return captureNone
}
return configured
}
// P8_ContentDefaultNone: sensitive fields MUST NOT be recorded unless mode is
// "full" and explicit opt-in was granted.
func recordsSensitiveContent(mode captureContentMode, optedIn bool) bool {
switch mode {
case captureNone, captureMetadata:
return false
case captureFull:
return optedIn // full requires explicit opt-in; without it, MUST be rejected
default:
return false
}
}
// endpointAttempt models one fan-out endpoint attempt result for §16.3.
type endpointAttempt struct {
url string
success bool
}
// attemptFanOut models §16.3: each of N endpoints attempted independently;
// one failure MUST NOT suppress attempts to others.
func attemptFanOut(endpoints []string, failing map[string]bool) []endpointAttempt {
results := make([]endpointAttempt, 0, len(endpoints))
for _, ep := range endpoints {
results = append(results, endpointAttempt{url: ep, success: !failing[ep]})
}
return results
}
// retryPolicy models §16.2 bounded retry.
type retryPolicy struct {
maxAttempts int
maxElapsed time.Duration
}
// isRetryPermitted implements INV1_RetryBounded: retry stops once either
// bound is exceeded; permanent failures are never retried indefinitely.
func (p retryPolicy) isRetryPermitted(attempt int, elapsed time.Duration, permanent bool) bool {
if permanent {
return false
}
if attempt >= p.maxAttempts {
return false
}
if elapsed >= p.maxElapsed {
return false
}
return true
}
// workflowFunctionalResult models §16.1 / Safeguards: observability failure
// MUST NOT change a successful functional workflow result into a failure.
type workflowFunctionalResult struct {
functionalSuccess bool
telemetryFailed bool
}
// P10_NonFatalExportFailure / SAFETY_FailClosedSecrets (functional-result
// component): the reported workflow result depends only on functional
// success, never on telemetry export outcome.
func (w workflowFunctionalResult) reportedSuccess() bool {
return w.functionalSuccess
}
// ---------------------------------------------------------------------------
// P1_OutcomeSeparateTrace, P3_OutcomeResultTaxonomy
// ---------------------------------------------------------------------------
func TestFormal_OutcomeEvaluationUsesSeparateTrace(t *testing.T) {
// §13.1: an outcome collector evaluating durable outputs long after
// workflow completion SHOULD create its own trace, not extend the
// original trace across hours/days.
tests := []struct {
name string
delaySinceRun time.Duration
extendsOldTrace bool
}{
{"immediate evaluation, same trace acceptable", 1 * time.Minute, true},
{"delayed evaluation, must use own trace", 26 * time.Hour, false},
{"multi-day delayed evaluation, must use own trace", 72 * time.Hour, false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
usesOwnTrace := tc.delaySinceRun > 24*time.Hour
if usesOwnTrace {
assert.False(t, tc.extendsOldTrace,
"outcome evaluation after long delay MUST NOT extend the original workflow trace (spec §13.1)")
}
})
}
}
func TestFormal_OutcomeResultTaxonomy(t *testing.T) {
// P3_OutcomeResultTaxonomy: gh-aw.outcome.result SHOULD use the canonical
// taxonomy defined in specs/safe-output-outcome-evaluation.md.
valid := validOutcomeTaxonomy()
cases := []struct {
name string
result outcomeResult
want bool
}{
{"accepted is valid", outcomeAccepted, true},
{"rejected is valid", outcomeRejected, true},
{"ignored is valid", outcomeIgnored, true},
{"pending is valid", outcomePending, true},
{"lifecycle is valid", outcomeLifecycle, true},
{"lifecycle_close is valid", outcomeLifecycleClose, true},
{"unknown value is invalid", outcomeResult("unknown_status"), false},
{"empty value is invalid", outcomeResult(""), false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
_, ok := valid[tc.result]
assert.Equal(t, tc.want, ok,
"gh-aw.outcome.result %q taxonomy membership mismatch (spec §13.3)", tc.result)
})
}
}
// ---------------------------------------------------------------------------
// P2_OutcomeSourceCorrelation, P9_MetricDimensionBound
// ---------------------------------------------------------------------------
func TestFormal_OutcomeSourceCorrelation(t *testing.T) {
// §13.2: when a full span context is unavailable, evaluation span SHOULD
// include source_run_id, source_workflow, and source_repo.
cases := []struct {
name string
span outcomeSpan
want bool
}{
{
name: "span link present satisfies correlation",
span: outcomeSpan{hasSourceLink: true},
want: true,
},
{
name: "full attribute triple satisfies correlation",
span: outcomeSpan{sourceRunID: "12345", sourceWorkflow: "daily.md", sourceRepo: "github/gh-aw"},
want: true,
},
{
name: "missing repo fails correlation",
span: outcomeSpan{sourceRunID: "12345", sourceWorkflow: "daily.md"},
want: false,
},
{
name: "no link and no attributes fails correlation",
span: outcomeSpan{},
want: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, tc.span.hasSourceCorrelation(),
"outcome span source correlation requirement mismatch (spec §13.2)")
})
}
}
func TestFormal_MetricDimensionCardinalityBound(t *testing.T) {
// P9_MetricDimensionBound: URLs and item identifiers MUST NOT be metric
// dimensions (§13.3, §15.4).
cases := []struct {
name string
dimensions []string
violates bool
}{
{"bounded dimensions only", []string{"outcome_type", "result"}, false},
{"url dimension violates bound", []string{"https://github.com/x/y/issues/1"}, true},
{"item_id dimension violates bound", []string{"item_id"}, true},
{"run_id dimension violates bound", []string{"source_run_id"}, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
span := outcomeSpan{metricDimensions: tc.dimensions}
assert.Equal(t, tc.violates, span.violatesMetricDimensionBound(),
"metric dimension cardinality bound check failed for %v (spec §13.3/§15.4)", tc.dimensions)
})
}
}
// ---------------------------------------------------------------------------
// P4_MirrorPathStable, P5_MirrorWriteBeforeExport, P6_MirrorNotTruncatedOnFail
// ---------------------------------------------------------------------------
func TestFormal_MirrorPathIsStable(t *testing.T) {
const expectedPath = "/tmp/gh-aw/otel.jsonl"
require.Equal(t, expectedPath, "/tmp/gh-aw/otel.jsonl",
"local telemetry mirror default path MUST remain %s (spec §14.1)", expectedPath)
}
func TestFormal_MirrorWriteOccursBeforeExportSuccess(t *testing.T) {
// P5_MirrorWriteBeforeExport: §14.3 "Mirror writes MUST occur before
// remote export success is assumed."
m := &mirrorWriter{}
record := `{"resourceSpans":[{}]}`
m.writeBeforeExport(record)
require.Len(t, m.written, 1, "mirror write MUST record the span before any export attempt (spec §14.3)")
assert.Equal(t, record, m.written[0], "mirror record content mismatch (spec §14.2)")
}
func TestFormal_MirrorNotTruncatedOnExportFailure(t *testing.T) {
// P6_MirrorNotTruncatedOnFail: a remote export failure MUST NOT delete or
// truncate previously written mirror data (spec §14.3).
cases := []struct {
name string
outcome exportOutcome
}{
{"export succeeds, mirror preserved", exportSuccess},
{"export fails, mirror still preserved", exportFailure},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
m := &mirrorWriter{}
m.writeBeforeExport(`{"resourceSpans":[{"a":1}]}`)
m.writeBeforeExport(`{"resourceSpans":[{"b":2}]}`)
after := m.simulateExport(tc.outcome)
assert.Len(t, after, 2,
"mirror data MUST survive export outcome %v without truncation (spec §14.3)", tc.outcome)
})
}
}
// ---------------------------------------------------------------------------
// P7_HeaderRedactionInMirror, P8_ContentDefaultNone
// ---------------------------------------------------------------------------
func TestFormal_HeaderRedactionBeforeMirrorOrDiagnostic(t *testing.T) {
// P7_HeaderRedactionInMirror: exporter headers/credentials MUST NOT
// appear in mirrored/artifact/diagnostic records (spec §15.1).
headers := map[string]string{
"Authorization": "Bearer super-secret-token",
"x-api-key": "abc123",
}
redacted := redactHeaders(headers)
require.Len(t, redacted, len(headers), "redaction MUST preserve header key set for observability without leaking values")
for k, v := range redacted {
assert.NotContains(t, v, "super-secret-token", "redacted header %q MUST NOT leak raw credential value (spec §15.1)", k)
assert.NotContains(t, v, "abc123", "redacted header %q MUST NOT leak raw credential value (spec §15.1)", k)
assert.Equal(t, "***", v, "redacted header value MUST be masked (spec §15.1)")
}
}
func TestFormal_ContentCaptureDefaultsToNone(t *testing.T) {
// P8_ContentDefaultNone: §5.6/§15.2 default capture-content MUST be
// "none"; raw prompts/responses/tool args/results MUST NOT be captured
// by default.
assert.Equal(t, captureNone, defaultCaptureMode(""),
"capture-content default MUST be 'none' when unset (spec §5.6)")
cases := []struct {
name string
mode captureContentMode
optedIn bool
captures bool
}{
{"none mode never captures", captureNone, false, false},
{"metadata mode never captures raw content", captureMetadata, false, false},
{"full mode without opt-in must not capture", captureFull, false, false},
{"full mode with explicit opt-in may capture", captureFull, true, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := recordsSensitiveContent(tc.mode, tc.optedIn)
assert.Equal(t, tc.captures, got,
"sensitive content capture decision mismatch for mode=%s optedIn=%v (spec §5.6, §15.2)", tc.mode, tc.optedIn)
})
}
}
// ---------------------------------------------------------------------------
// P11_PartialFanOutIndependent, INV1_RetryBounded
// ---------------------------------------------------------------------------
func TestFormal_PartialFanOutFailureDoesNotSuppressOthers(t *testing.T) {
// P11_PartialFanOutIndependent: §16.3 for N endpoints, each MUST be
// attempted independently; partial success MUST NOT be discarded.
endpoints := []string{"(a.example/redacted), "(b.example/redacted), "(c.example/redacted)
failing := map[string]bool{"(b.example/redacted) true}
results := attemptFanOut(endpoints, failing)
require.Len(t, results, len(endpoints), "all configured endpoints MUST be attempted (spec §16.3)")
successCount := 0
for _, r := range results {
if r.success {
successCount++
}
}
assert.Equal(t, 2, successCount,
"one endpoint failure MUST NOT suppress attempts or successes at other endpoints (spec §16.3)")
// verify the specific failing endpoint's failure is isolated
for _, r := range results {
if r.url == "(b.example/redacted) {
assert.False(t, r.success, "the deliberately failing endpoint MUST be reported as failed")
} else {
assert.True(t, r.success, "endpoint %s unaffected by sibling failure MUST report success", r.url)
}
}
}
func TestFormal_RetryIsBoundedByAttemptsAndElapsedTime(t *testing.T) {
// INV1_RetryBounded: §16.2 retry MUST be bounded by max attempts and max
// elapsed time; permanent failures MUST NOT be retried indefinitely.
policy := retryPolicy{maxAttempts: 3, maxElapsed: 10 * time.Second}
cases := []struct {
name string
attempt int
elapsed time.Duration
permanent bool
permitted bool
}{
{"first attempt within bounds is permitted", 0, 1 * time.Second, false, true},
{"attempt at max count is denied", 3, 1 * time.Second, false, false},
{"attempt within count but elapsed exceeded is denied", 1, 11 * time.Second, false, false},
{"permanent failure is never retried regardless of bounds", 0, 0, true, false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := policy.isRetryPermitted(tc.attempt, tc.elapsed, tc.permanent)
assert.Equal(t, tc.permitted, got,
"retry permission mismatch for attempt=%d elapsed=%v permanent=%v (spec §16.2)", tc.attempt, tc.elapsed, tc.permanent)
})
}
}
// ---------------------------------------------------------------------------
// P10_NonFatalExportFailure / SAFETY_FailClosedSecrets
// ---------------------------------------------------------------------------
func TestFormal_TelemetryFailureNeverFlipsFunctionalSuccess(t *testing.T) {
// P10_NonFatalExportFailure & the "Safeguards" block: observability
// failure SHOULD NOT change a successful functional workflow into a
// failed workflow. Reported result depends only on functional outcome.
cases := []struct {
name string
functionalSuccess bool
telemetryFailed bool
wantReported bool
}{
{"functional success, telemetry ok", true, false, true},
{"functional success, telemetry failed", true, true, true},
{"functional failure, telemetry ok", false, false, false},
{"functional failure, telemetry also failed", false, true, false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
w := workflowFunctionalResult{functionalSuccess: tc.functionalSuccess, telemetryFailed: tc.telemetryFailed}
assert.Equal(t, tc.wantReported, w.reportedSuccess(),
"reported workflow result MUST depend only on functional success, not telemetry outcome (spec §16.1, Safeguards)")
})
}
}
// TestFormal_SafeguardsShutdownDoesNotDeleteMirrorOnTimeout is an edge case
// covering the Safeguards block: a bounded exporter flush timeout at
// shutdown MUST NOT delete already-written mirror records.
func TestFormal_SafeguardsShutdownDoesNotDeleteMirrorOnTimeout(t *testing.T) {
m := &mirrorWriter{}
m.writeBeforeExport(`{"resourceSpans":[{"final":true}]}`)
// simulate a bounded flush timing out during shutdown
flushTimedOut := true
survivingRecords := m.simulateExport(exportFailure)
require.True(t, flushTimedOut, "test setup: shutdown flush must simulate a timeout")
assert.Len(t, survivingRecords, 1,
"shutdown flush timeout MUST NOT delete previously written mirror records (spec Safeguards)")
}
Caution
agentic threat detected
Threat detection flagged this output in warn mode. Manual review is REQUIRED before any follow-up automation.
Details
Potential security threats were detected in the agent output.
Review the workflow run logs for details.
Summary
Formalized Sections 13-16 of the gh-aw OpenTelemetry Observability Specification (
specs/otel-observability-spec.md, v0.4.0): Outcome Evaluation, Local Mirrors and Artifacts, Security and Privacy, and Reliability and Failure Handling. These sections describe the outcome-evaluation span/trace-correlation contract, the/tmp/gh-aw/otel.jsonllocal mirror durability guarantees, header/content redaction defaults, and bounded-retry/fail-closed reliability behavior. No concrete Go implementation of the outcome-evaluation span emitter or mirror writer exists yet inpkg/workflow, so the generated test suite models the behavior against minimal stub interfaces derived directly from the normative MUST/SHOULD statements, ready to be wired to real call sites once implemented.Specification
specs/otel-observability-spec.mdFormal Model
Predicates and invariants (illustrative notation)
Behavioral Coverage Map
P1_OutcomeSeparateTraceTestFormal_OutcomeEvaluationUsesSeparateTraceP2_OutcomeSourceCorrelationTestFormal_OutcomeSourceCorrelationP3_OutcomeResultTaxonomyTestFormal_OutcomeResultTaxonomygh-aw.outcome.resultrestricted to canonical taxonomy valuesP9_MetricDimensionBoundTestFormal_MetricDimensionCardinalityBoundP4_MirrorPathStableTestFormal_MirrorPathIsStable/tmp/gh-aw/otel.jsonlP5_MirrorWriteBeforeExportTestFormal_MirrorWriteOccursBeforeExportSuccessP6_MirrorNotTruncatedOnFailTestFormal_MirrorNotTruncatedOnExportFailureP7_HeaderRedactionInMirrorTestFormal_HeaderRedactionBeforeMirrorOrDiagnosticP8_ContentDefaultNoneTestFormal_ContentCaptureDefaultsToNonenone;fullmode requires explicit opt-in to captureP11_PartialFanOutIndependentTestFormal_PartialFanOutFailureDoesNotSuppressOthersINV1_RetryBoundedTestFormal_RetryIsBoundedByAttemptsAndElapsedTimeP10_NonFatalExportFailureTestFormal_TelemetryFailureNeverFlipsFunctionalSuccessSAFETY_FailClosedSecrets(edge case)TestFormal_SafeguardsShutdownDoesNotDeleteMirrorOnTimeoutGenerated Test Suite
📄
pkg/workflow/otel_reliability_formal_test.goUsage
pkg/workflow/otel_reliability_formal_test.go(already placed there during formalization).outcomeSpan,mirrorWriter,retryPolicy,workflowFunctionalResult,redactHeaders, etc.) with real call sites once the outcome-evaluation span emitter, mirror writer, and fan-out/retry orchestration are implemented inpkg/workflow.go test ./pkg/workflow/... -run TestFormal_Context
specs/otel-observability-spec.md(Sections 13-16; prior runs on 2026-08-02 covered Sections 4-12 and are reflected inpkg/workflow/otel_observability_formal_test.go)