Source review and production activation have separate evidence; the status here records what the corresponding live canary actually proved.
The first full-integration canary for this historical source failed. Reviewed repairs later passed bounded gates, and OutcomeReady now runs continuously in production on the reviewed build. Shown at the current implementation head rather than the canary revision.
- Failed promotion gate The first OutcomeReady canary exposed notification contention, false correlation, observer misclassification, and missing producer coverage.
- Notifier repair Independent exact-head review passed after the mail writer was serialized and bounded diagnostics were made UTF-8 safe.
- Successful bounded canary The repaired run delivered five stranded outcomes and one Temporal-origin formula result with exact Workflow and Run correlation.
- Nine outcomes acknowledged The coordinator verified nine unique outcomes as acknowledged; repeated attention mail came from later redelivery cycles.
- Returned to shadow The worker finished the bounded recovery in shadow mode for both agent mutation and outcome delivery.
- Bounded finalization The installed operator finalized the first production work item with an atomic receipt while preserving its acknowledged outbox; the worker stayed in shadow mode.
- Exact-five producer canary The all-store preflight found only five authorized transitions. All five produced first-cycle deliveries, then committed exact coordinator acknowledgements in a bounded acknowledgement phase.
- Atomic finalization The reviewed operator finalized the remaining work item without creating another outcome generation; the worker returned to shadow mode with an empty health surface.
- Continuous OutcomeReady activation After a separate approval, the reviewed worker was left running continuously, with agent mutation still in shadow and only the result-delivery path active. The running binary matched the reviewed build.
- Production receipts verified The first two production outcomes, one from a project store and one from the city store, were delivered and acknowledged on their first cycle under a single coordinator fence.
- Postconditions held A later synthetic close was durably excluded as a non-outcome because its source result had already been acknowledged and finalized. Operators then disarmed the rollback timer with the same process and configuration still running, and the all-store health surface stayed empty.
How to read this file
Follow the execution boundary
Click any line section to open its explanation. The complete file stays visible, and the selected explanation opens directly beneath the code it describes.
package temporalbeads
import (
"fmt"
"time"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
)
const (
CoordinatorOutcomeTaskQueue = "gascity-coordinator-outcomes"
CoordinatorOutcomeWorkflowName = "CoordinatorOutcomeWorkflow"
SignalOutcomeReady = "coordinator.outcome-ready"
SignalCoordinatorAcknowledged = "coordinator.acknowledged"
QueryCoordinatorOutcomeState = "coordinator-outcome-state"
defaultOutcomeRedeliveryPeriod = time.Minute
defaultOutcomeContinueAsNewAfter = 100
outcomeAckAfterDeliveryFailureChangeID = "coordinator-outcome-ack-after-delivery-failure"
outcomeAckAfterDeliveryFailureVersion = 1
)
type CoordinatorOutcomeInput struct {
Envelope OutcomeReady `json:"envelope"`
RedeliveryInterval time.Duration `json:"redelivery_interval"`
ContinueAsNewAfter int `json:"continue_as_new_after,omitempty"`
ResumeState *CoordinatorOutcomeState `json:"resume_state,omitempty"`
}
// CoordinatorOutcomeState remains queryable at needs-coordinator-ack until the
// exact current delivery fence is explicitly acknowledged.
type CoordinatorOutcomeState struct {
Envelope OutcomeReady `json:"envelope"`
Phase OutcomeCoordinatorState `json:"phase"`
DeliveryAttempts int `json:"delivery_attempts"`
DuplicateSignals int `json:"duplicate_signals"`
StaleAcknowledgements int `json:"stale_acknowledgements"`
DeliveryRef string `json:"delivery_ref,omitempty"`
CoordinatorFence string `json:"coordinator_fence,omitempty"`
DeliveredAt time.Time `json:"delivered_at,omitempty"`
AcknowledgedAt time.Time `json:"acknowledged_at,omitempty"`
LastDeliveryError string `json:"last_delivery_error,omitempty"`
ContinueAsNewCount int `json:"continue_as_new_count"`
}
func CoordinatorOutcomeWorkflowID(envelope OutcomeReady) (string, error) {
if err := envelope.Validate(); err != nil {
return "", err
}
return CoordinatorOutcomeWorkflowIDForOutcome(envelope.OutcomeID)
}
func CoordinatorOutcomeWorkflowIDForOutcome(outcomeID string) (string, error) {
if err := validateSegment("coordinator outcome id", outcomeID); err != nil {
return "", err
}
workflowID := "coordinator-outcome/" + outcomeID
if err := validateWorkflowID("coordinator outcome workflow id", workflowID); err != nil {
return "", err
}
return workflowID, nil
}Lines 1–63
Give every result a stable execution
- What this section does
- Defines the OutcomeReady Workflow contract, queryable state, and stable Workflow ID derived from the immutable outcome ID.
- Why it matters for Temporal
- The result is represented by a durable execution rather than a best-effort message that can disappear when the coordinator is unavailable.
- Stable Workflow ID
- Queryable phase
- Immutable outcome
// CoordinatorOutcomeWorkflow owns retryable local delivery and the explicit
// acknowledgement gate. A coordinator restart changes the delivery fence on a
// later redelivery; stale acknowledgements cannot complete the Workflow.
func CoordinatorOutcomeWorkflow(
ctx workflow.Context,
input CoordinatorOutcomeInput,
) (CoordinatorOutcomeState, error) {
if err := input.Envelope.Validate(); err != nil {
return CoordinatorOutcomeState{}, temporal.NewNonRetryableApplicationError(
"invalid coordinator outcome input",
"InvalidOutcomeReady",
err,
)
}
if input.RedeliveryInterval <= 0 {
input.RedeliveryInterval = defaultOutcomeRedeliveryPeriod
}
if input.ContinueAsNewAfter <= 0 {
input.ContinueAsNewAfter = defaultOutcomeContinueAsNewAfter
}
state := CoordinatorOutcomeState{
Envelope: cloneOutcomeReady(input.Envelope),
Phase: OutcomeCoordinatorPending,
}
if input.ResumeState != nil {
if !outcomesEqual(input.Envelope, input.ResumeState.Envelope) {
return CoordinatorOutcomeState{}, temporal.NewNonRetryableApplicationError(
"continued coordinator outcome changed identity",
"InvalidOutcomeContinuation",
nil,
)
}
state = *input.ResumeState
state.Envelope = cloneOutcomeReady(input.ResumeState.Envelope)
}
if err := workflow.SetQueryHandler(
ctx,
QueryCoordinatorOutcomeState,
func() (CoordinatorOutcomeState, error) { return state, nil },
); err != nil {
return state, err
}
readyCh := workflow.GetSignalChannel(ctx, SignalOutcomeReady)
ackCh := workflow.GetSignalChannel(ctx, SignalCoordinatorAcknowledged)
runDeliveryAttempts := 0
if input.ResumeState != nil {
for {
var duplicate OutcomeReady
if !readyCh.ReceiveAsync(&duplicate) {
break
}
if outcomesEqual(state.Envelope, duplicate) {
state.DuplicateSignals++
} else {
state.LastDeliveryError = "conflicting-outcome-ready-signal"
}
}
for {
var boundaryAck OutcomeAcknowledgement
if !ackCh.ReceiveAsync(&boundaryAck) {
break
}
if !outcomeAcknowledgementMatches(state, boundaryAck) {
state.StaleAcknowledgements++
continue
}
record, err := executeOutcomeAcknowledgement(
ctx,
boundaryAck,
state.DeliveryAttempts,
)
if err != nil {
state.LastDeliveryError = "canonical-acknowledgement-failed"
break
}
recordOutcomeAcknowledgement(ctx, &state, record)
return state, nil
}
}Lines 64–143
Rebuild state without accepting stale evidence
- What this section does
- Validates the envelope, restores Continue-As-New state, drains buffered duplicate signals, and rejects acknowledgements for the wrong delivery fence.
- Why it matters for Temporal
- Replay and Continue-As-New reconstruct the procedure while the application-level fence prevents an old coordinator session from completing a newer delivery.
- Continue-As-New
- Duplicate signal handling
- Stale acknowledgement fence
for {
state.DeliveryAttempts++
runDeliveryAttempts++
delivery, err := executeOutcomeDelivery(
ctx,
state.Envelope,
state.DeliveryAttempts,
)
if err != nil {
state.LastDeliveryError = "local-delivery-failed"
acknowledged, err := awaitAcknowledgementAfterDeliveryFailure(
ctx,
readyCh,
ackCh,
input.RedeliveryInterval,
&state,
)
if err != nil {
return state, err
}
if acknowledged {
return state, nil
}
if runDeliveryAttempts >= input.ContinueAsNewAfter {
return continueCoordinatorOutcomeAsNew(ctx, input, state)
}
continue
}
state.Phase = OutcomeCoordinatorNeedsAck
state.DeliveryRef = delivery.DeliveryRef
state.CoordinatorFence = delivery.CoordinatorFence
if state.DeliveredAt.IsZero() {
state.DeliveredAt = workflow.Now(ctx)
}
state.LastDeliveryError = ""
redeliver, acknowledgement, err := awaitOutcomeAcknowledgement(
ctx,
readyCh,
ackCh,
input.RedeliveryInterval,
&state,
)
if err != nil {
return state, err
}
if redeliver {
if runDeliveryAttempts >= input.ContinueAsNewAfter {
return continueCoordinatorOutcomeAsNew(ctx, input, state)
}
continue
}
record, err := executeOutcomeAcknowledgement(
ctx,
acknowledgement,
state.DeliveryAttempts,
)
if err != nil {
state.LastDeliveryError = "canonical-acknowledgement-failed"
if err := workflow.Sleep(ctx, input.RedeliveryInterval); err != nil {
return state, err
}
if runDeliveryAttempts >= input.ContinueAsNewAfter {
return continueCoordinatorOutcomeAsNew(ctx, input, state)
}
continue
}
recordOutcomeAcknowledgement(ctx, &state, record)
return state, nil
}
}Lines 144–215
Redeliver until the coordinator proves receipt
- What this section does
- Runs the notification Activity, exposes needs-ack state, waits for the exact acknowledgement, and retries both delivery and canonical acknowledgement failures.
- Why it matters for Temporal
- A queued notification is attention, not completion. The Workflow remains open until the canonical store records acknowledgement.
- Durable retry
- Needs-ack phase
- Explicit completion
func awaitAcknowledgementAfterDeliveryFailure(
ctx workflow.Context,
readyCh workflow.ReceiveChannel,
ackCh workflow.ReceiveChannel,
redeliveryInterval time.Duration,
state *CoordinatorOutcomeState,
) (bool, error) {
patched := workflow.GetVersion(
ctx,
outcomeAckAfterDeliveryFailureChangeID,
workflow.DefaultVersion,
outcomeAckAfterDeliveryFailureVersion,
) != workflow.DefaultVersion
if !patched || state.Phase != OutcomeCoordinatorNeedsAck {
return false, workflow.Sleep(ctx, redeliveryInterval)
}
redeliver, acknowledgement, err := awaitOutcomeAcknowledgement(
ctx,
readyCh,
ackCh,
redeliveryInterval,
state,
)
if err != nil || redeliver {
return false, err
}
record, err := executeOutcomeAcknowledgement(
ctx,
acknowledgement,
state.DeliveryAttempts,
)
if err != nil {
state.LastDeliveryError = "canonical-acknowledgement-failed"
return false, workflow.Sleep(ctx, redeliveryInterval)
}
recordOutcomeAcknowledgement(ctx, state, record)
return true, nil
}Lines 216–254
Patch the failed-delivery wait behind a version marker
- What this section does
- Uses workflow.GetVersion to gate a patched wait: when a redelivery fails after the outcome already reached needs-ack, the Workflow watches for the exact acknowledgement during the pause and commits it instead of only sleeping.
- Why it matters for Temporal
- GetVersion records the branch in Event History, so executions started before the patch replay the original sleep while new ones take the acknowledgement-aware path without nondeterminism errors.
- workflow.GetVersion
- Ack during failure wait
- Replay-safe patch
func recordOutcomeAcknowledgement(
ctx workflow.Context,
state *CoordinatorOutcomeState,
record OutcomeRecord,
) {
state.Phase = OutcomeCoordinatorAcknowledged
state.AcknowledgedAt = record.AcknowledgedAt
if state.AcknowledgedAt.IsZero() {
state.AcknowledgedAt = workflow.Now(ctx)
}
}
func continueCoordinatorOutcomeAsNew(
ctx workflow.Context,
input CoordinatorOutcomeInput,
state CoordinatorOutcomeState,
) (CoordinatorOutcomeState, error) {
state.ContinueAsNewCount++
resume := state
resume.Envelope = cloneOutcomeReady(state.Envelope)
input.ResumeState = &resume
return state, workflow.NewContinueAsNewError(
ctx,
CoordinatorOutcomeWorkflowName,
input,
)
}
func executeOutcomeDelivery(
ctx workflow.Context,
envelope OutcomeReady,
attempt int,
) (OutcomeDelivery, error) {
deliveryCycle := fmt.Sprintf("cycle-%06d", attempt)
options := workflow.ActivityOptions{
ActivityID: fmt.Sprintf("deliver/%s/%s", envelope.OutcomeID, deliveryCycle),
StartToCloseTimeout: 30 * time.Second,
ScheduleToCloseTimeout: 2 * time.Minute,
RetryPolicy: &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2,
MaximumInterval: 10 * time.Second,
MaximumAttempts: 5,
},
}
var delivery OutcomeDelivery
err := workflow.ExecuteActivity(
workflow.WithActivityOptions(ctx, options),
DeliverCoordinatorOutcomeActivityName,
OutcomeDeliveryRequest{
Envelope: envelope, DeliveryCycle: deliveryCycle,
},
).Get(ctx, &delivery)
if err != nil {
return OutcomeDelivery{}, err
}
if err := validateOutcomeDelivery(delivery); err != nil {
return OutcomeDelivery{}, temporal.NewNonRetryableApplicationError(
"invalid coordinator delivery receipt",
"InvalidOutcomeDelivery",
err,
)
}
if delivery.OutcomeID != envelope.OutcomeID ||
delivery.WorkID != envelope.WorkID {
return OutcomeDelivery{}, temporal.NewNonRetryableApplicationError(
"coordinator delivery receipt changed outcome identity",
"InvalidOutcomeDelivery",
nil,
)
}
return delivery, nil
}Lines 255–328
Bound history and identify each delivery cycle
- What this section does
- Records the acknowledged phase and timestamp, carries state through Continue-As-New, and assigns a stable Activity ID to each logical redelivery cycle.
- Why it matters for Temporal
- Activity retries within one cycle can deduplicate their side effects, while a later cycle can create fresh attention without growing one history forever.
- Cycle identity
- Activity retry
- Bounded history
func awaitOutcomeAcknowledgement(
ctx workflow.Context,
readyCh workflow.ReceiveChannel,
ackCh workflow.ReceiveChannel,
redeliveryInterval time.Duration,
state *CoordinatorOutcomeState,
) (bool, OutcomeAcknowledgement, error) {
timerCtx, cancelTimer := workflow.WithCancel(ctx)
defer cancelTimer()
timer := workflow.NewTimer(timerCtx, redeliveryInterval)
for {
var acknowledgement OutcomeAcknowledgement
var ackReceived, redeliver bool
selector := workflow.NewSelector(ctx)
selector.AddReceive(readyCh, func(channel workflow.ReceiveChannel, _ bool) {
var duplicate OutcomeReady
channel.Receive(ctx, &duplicate)
if outcomesEqual(state.Envelope, duplicate) {
state.DuplicateSignals++
return
}
state.LastDeliveryError = "conflicting-outcome-ready-signal"
})
selector.AddReceive(ackCh, func(channel workflow.ReceiveChannel, _ bool) {
channel.Receive(ctx, &acknowledgement)
if !outcomeAcknowledgementMatches(*state, acknowledgement) {
state.StaleAcknowledgements++
return
}
ackReceived = true
})
selector.AddFuture(timer, func(workflow.Future) { redeliver = true })
selector.Select(ctx)
if ackReceived {
cancelTimer()
return false, acknowledgement, nil
}
if redeliver {
for {
var boundaryAck OutcomeAcknowledgement
if !ackCh.ReceiveAsync(&boundaryAck) {
break
}
if !outcomeAcknowledgementMatches(*state, boundaryAck) {
state.StaleAcknowledgements++
continue
}
cancelTimer()
return false, boundaryAck, nil
}
for {
var duplicate OutcomeReady
if !readyCh.ReceiveAsync(&duplicate) {
break
}
if outcomesEqual(state.Envelope, duplicate) {
state.DuplicateSignals++
} else {
state.LastDeliveryError = "conflicting-outcome-ready-signal"
}
}
return true, OutcomeAcknowledgement{}, nil
}
}
}
func outcomeAcknowledgementMatches(
state CoordinatorOutcomeState,
acknowledgement OutcomeAcknowledgement,
) bool {
return validateOutcomeAcknowledgement(acknowledgement) == nil &&
acknowledgement.StoreRef == state.Envelope.StoreRef &&
acknowledgement.OutcomeID == state.Envelope.OutcomeID &&
acknowledgement.WorkID == state.Envelope.WorkID &&
acknowledgement.CoordinatorFence == state.CoordinatorFence
}Lines 329–405
Wait on signals and a durable timer
- What this section does
- Selects between duplicate outcome signals, exact acknowledgement, and the redelivery timer while counting stale or conflicting input.
- Why it matters for Temporal
- The wait survives Worker loss and remains inspectable in Event History; no polling process has to remember when to try again.
- Selector
- Durable timer
- Exact tuple matching
func executeOutcomeAcknowledgement(
ctx workflow.Context,
ack OutcomeAcknowledgement,
deliveryCycle int,
) (OutcomeRecord, error) {
ackCycle := fmt.Sprintf("cycle-%06d", deliveryCycle)
options := workflow.ActivityOptions{
ActivityID: "ack/" + ack.OutcomeID + "/" + ackCycle,
StartToCloseTimeout: 30 * time.Second,
ScheduleToCloseTimeout: 2 * time.Minute,
RetryPolicy: &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2,
MaximumInterval: 10 * time.Second,
MaximumAttempts: 5,
},
}
var record OutcomeRecord
err := workflow.ExecuteActivity(
workflow.WithActivityOptions(ctx, options),
AcknowledgeCoordinatorOutcomeActivityName,
ack,
).Get(ctx, &record)
return record, err
}Lines 406–431
Commit acknowledgement through a separate Activity
- What this section does
- Writes the acknowledgement to the canonical store with a cycle-specific Activity ID and bounded retry policy.
- Why it matters for Temporal
- The Workflow only completes after the nondeterministic store mutation returns a durable receipt.
- Canonical receipt
- Unique Activity ID
- Bounded retry