activities

package
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 23, 2026 License: MIT Imports: 7 Imported by: 0

Documentation

Overview

Package activities hosts the durable activities for the go-workflows pilot: the observe-only shadow activities (Phase 3) and the test-only execution activities (Phase 4). Activities touch the current system only through the narrow ports defined in internal/projectdurable. The Phase 3 observe activities write nothing except shadow-comparison metadata; the Phase 4 execution activities additionally drive the current service's claim/complete contract through ports, but only test adapters implement those ports until cutover approval - no production path constructs them.

Shared workflow DTOs (AutomationRunWorkflowInput, ShadowRunTrace) live in this package rather than in the workflows package: the WriteShadowComparison activity takes the trace as input, so placing the types here lets workflows import activities (one clean direction) without an import cycle. The workflows package re-exports them via type aliases.

Index

Constants

View Source 
const (
	ActivityLoadRunSnapshot      = "load-run-snapshot"
	ActivityValidateExecutable   = "validate-run-executable"
	ActivityObserveWorkTask      = "observe-work-task"
	ActivityObserveCodexAttempt  = "observe-codex-attempt"
	ActivityObserveCloseout      = "observe-governed-closeout"
	ActivityObserveReviewQueue   = "observe-review-queue"
	ActivityObserveVerifierState = "observe-verifier-state"
	ActivityWriteShadow          = "write-shadow-comparison"
)

Activity step names recorded in durable traces and shadow comparisons. They must pass projectdurable.ValidateSafeRef.

View Source 
const (
	ActivityClaimRunForExecution     = "claim-run-for-execution"
	ActivityCompleteRunAttempt       = "complete-run-attempt"
	ActivityWriteExecutionComparison = "write-execution-comparison"
)

Activity step names for the Phase 4 test-only execution workflow.

View Source 
const (
	ActivityResolveChainStages         = "resolve-chain-stages"
	ActivityCompileChainStage          = "compile-chain-stage"
	ActivityReleaseCompiledTasks       = "release-compiled-tasks"
	ActivityActivateStagePlan          = "activate-stage-plan"
	ActivityObserveStagePlan           = "observe-stage-plan"
	ActivityCarryForwardOutputs        = "carry-forward-outputs"
	ActivityObserveChainGitOps         = "observe-chain-gitops"
	ActivityWriteChainShadowComparison = "write-chain-shadow-comparison"
)
View Source 
const (
	ChainTraceStatusRunning   = "running"
	ChainTraceStatusCompleted = "completed"
	ChainTraceStatusBlocked   = "blocked"
	ChainTraceStatusFailed    = "failed"
)
View Source 
const (
	ChainFailureStageObservationTimeout = "stage-observation-timeout"
	ChainFailureStageBlocked            = "stage-blocked"
	ChainFailureStageFailed             = "stage-failed"
	ChainFailureActivityFailed          = "activity-failed"
	ChainFailureGitOpsNotReady          = "gitops-not-ready"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type AutomationRunActivities 

type AutomationRunActivities struct {
	Runs   projectdurable.AutomationRunObserver
	Tasks  projectdurable.WorkTaskObserver
	Shadow projectdurable.ShadowComparisonWriter
}

AutomationRunActivities groups the observe-only activities and their ports. Register one instance per worker; methods are stateless beyond the injected ports and safe for concurrent execution.

func (*AutomationRunActivities) LoadAutomationRunSnapshot 

func (a *AutomationRunActivities) LoadAutomationRunSnapshot(ctx context.Context, ref projectdurable.SafeAutomationRunRef) (projectdurable.DurableRunSnapshot, error)

LoadAutomationRunSnapshot loads the metadata-only run snapshot through the Runs port. Both the input ref and the returned snapshot are validated fail-closed before entering durable history.

func (*AutomationRunActivities) ObserveCodexAttempt 

func (a *AutomationRunActivities) ObserveCodexAttempt(ctx context.Context, snap projectdurable.DurableRunSnapshot) (projectdurable.DurableActivityResult, error)

ObserveCodexAttempt classifies the run's attempt outcome from the snapshot into safe categories only. A snapshot carrying a safe failure category (timeout, codex auth unavailable, schema invalid, ...) classifies blocked with that category; completed/verifying classify ok with a verifying expectation; in-flight statuses classify ok; anything else fails closed to blocked without inventing a category.

func (*AutomationRunActivities) ObserveGovernedCloseout 

func (a *AutomationRunActivities) ObserveGovernedCloseout(ctx context.Context, snap projectdurable.DurableRunSnapshot) (projectdurable.DurableActivityResult, error)

ObserveGovernedCloseout is metadata-only: it reports the closeout-related ref counts present on the snapshot (evidence, claim, knowledge-candidate refs). It never performs a closeout.

func (*AutomationRunActivities) ObserveReviewQueue 

func (a *AutomationRunActivities) ObserveReviewQueue(ctx context.Context, snap projectdurable.DurableRunSnapshot) (projectdurable.DurableActivityResult, error)

ObserveReviewQueue reports review-result ref presence. Convention (documented per the Phase 3 spec): review is REQUIRED exactly when the snapshot status is "verifying" - the verifying parking state is the review/verification gate in the current run contract - and the requirement is satisfied via snapshot refs (ReviewResultRefs non-empty). Required but absent classifies blocked with FailureCategoryVerificationRequired.

func (*AutomationRunActivities) ObserveVerifierState 

func (a *AutomationRunActivities) ObserveVerifierState(ctx context.Context, snap projectdurable.DurableRunSnapshot) (projectdurable.DurableActivityResult, error)

ObserveVerifierState classifies ok only when verifier result refs exist on the snapshot. Otherwise it classifies blocked with FailureCategoryVerificationRequired: a run with status verifying and no verifier refs must never classify as complete.

func (*AutomationRunActivities) ObserveWorkTask 

func (a *AutomationRunActivities) ObserveWorkTask(ctx context.Context, ref projectdurable.SafeWorkTaskRef) (projectdurable.DurableActivityResult, error)

ObserveWorkTask reads the work-task snapshot through the Tasks port and reports the task status plus bounded ref counts. Observe-only: no claiming in this phase (the plan's ObserveOrClaimWorkTask claim path is Phase 4). Result refs carry "task-status:<status>" and one "<kind>:<count>" entry per ref kind, with underscores dashed (e.g. "verifier-result-refs:2").

func (*AutomationRunActivities) ValidateAutomationRunExecutable 

func (a *AutomationRunActivities) ValidateAutomationRunExecutable(ctx context.Context, snap projectdurable.DurableRunSnapshot) (projectdurable.DurableActivityResult, error)

ValidateAutomationRunExecutable is a pure classification activity: it maps the snapshot status and failure category to an activity result without touching any port. Executable-in-shadow statuses (queued, claiming, starting, running, verifying) classify ok; any snapshot carrying a safe failure category - or a terminal/unknown status - classifies blocked.

func (*AutomationRunActivities) WriteShadowComparison 

func (a *AutomationRunActivities) WriteShadowComparison(ctx context.Context, ref projectdurable.SafeAutomationRunRef, trace ShadowRunTrace) (projectdurable.DurableActivityResult, error)

WriteShadowComparison flattens the trace into bounded key->value fields (every key passes ValidateSafeRef, every value passes ValidateSafeSummary; keys are validated in sorted order for deterministic failures) and writes them through the Shadow port. This is the only write in the shadow phase.

type AutomationRunExecutionActivities 

type AutomationRunExecutionActivities struct {
	Claim    projectdurable.WorkTaskClaimPort
	Complete projectdurable.AttemptCompletionPort
	Runs     projectdurable.AutomationRunObserver
	Shadow   projectdurable.ShadowComparisonWriter
}

AutomationRunExecutionActivities groups the Phase 4 test-only execution activities and their ports. Implementations of Claim and Complete are test adapters over the CURRENT projectautomation service until cutover approval; durable code never imports that service. Runs is used for the durable post-completion snapshot re-read; Shadow records the execution comparison. Register one instance per worker alongside AutomationRunActivities (whose observe methods the execution workflow reuses instead of duplicating).

func (*AutomationRunExecutionActivities) ClaimRunForExecution 

func (a *AutomationRunExecutionActivities) ClaimRunForExecution(ctx context.Context, ref projectdurable.SafeAutomationRunRef, runnerID string) (ClaimedRunResult, error)

ClaimRunForExecution claims the referenced run through the Claim port on behalf of runnerID. Everything is validated fail-closed: the input ref and runner id before the port call, the returned snapshot and claim token after it. Port errors are genericized so no port error text enters durable history.

func (*AutomationRunExecutionActivities) CompleteRunAttempt 

func (a *AutomationRunExecutionActivities) CompleteRunAttempt(ctx context.Context, ref projectdurable.SafeAutomationRunRef, outcome projectdurable.DurableAttemptOutcome) (projectdurable.DurableRunSnapshot, error)

CompleteRunAttempt reports one attempt outcome through the Complete port. The outcome is validated fail-closed BEFORE the port is invoked: an outcome carrying unsafe refs (roots, markers, URLs, emails, ...) must never reach the current service. When the Runs observer is configured, the post-completion snapshot is re-read through it (mirroring the current service's durable re-read after completion) and the re-read wins.

func (*AutomationRunExecutionActivities) WriteExecutionComparison 

func (a *AutomationRunExecutionActivities) WriteExecutionComparison(ctx context.Context, ref projectdurable.SafeAutomationRunRef, trace ExecutionTrace) (projectdurable.DurableActivityResult, error)

WriteExecutionComparison flattens the execution trace into bounded key->value fields and writes them through the Shadow port, exactly like the Phase 3 WriteShadowComparison (keys validated in sorted order for deterministic failures). This is the only Shadow write in the execution workflow and it always runs last.

type AutomationRunWorkflowInput 

type AutomationRunWorkflowInput struct {
	ProjectID    string `json:"project_id"`
	AutomationID string `json:"automation_id"`
	RunID        string `json:"run_id"`
	TaskID       string `json:"task_id,omitempty"`
	TraceID      string `json:"trace_id,omitempty"`
	ShadowOnly   bool   `json:"shadow_only"`
}

AutomationRunWorkflowInput is the metadata-only input of the shadow workflow. JSON-serializable ids and a shadow flag - never prompts, source, roots, or secrets.

type CarriedTaskIDs 

type CarriedTaskIDs struct {
	TaskIDs []string `json:"task_ids,omitempty"`
}

type ChainGitOpsObservation 

type ChainGitOpsObservation struct {
	GitOpsReady    bool              `json:"gitops_ready"`
	PullRequestRef string            `json:"pull_request_ref,omitempty"`
	RecoveryStatus string            `json:"recovery_status,omitempty"`
	BlockedReason  string            `json:"blocked_reason,omitempty"`
	Metadata       map[string]string `json:"metadata,omitempty"`
}

type ChainShadowTrace 

type ChainShadowTrace struct {
	Input           ChainShadowWorkflowInput               `json:"input"`
	Stages          []StageTraceEntry                      `json:"stages,omitempty"`
	Steps           []projectdurable.DurableActivityResult `json:"steps,omitempty"`
	FinalStatus     string                                 `json:"final_status"`
	GitOpsReady     bool                                   `json:"gitops_ready"`
	PullRequestRef  string                                 `json:"pull_request_ref,omitempty"`
	GitOpsMetadata  map[string]string                      `json:"gitops_metadata,omitempty"`
	FailureCategory string                                 `json:"failure_category,omitempty"`
}

type ChainShadowWorkflowInput 

type ChainShadowWorkflowInput struct {
	ProjectID                string `json:"project_id"`
	ChainRef                 string `json:"chain_ref"`
	InputRef                 string `json:"input_ref"`
	TraceID                  string `json:"trace_id,omitempty"`
	ShadowOnly               bool   `json:"shadow_only"`
	MaxStageObservationPolls int    `json:"max_stage_observation_polls,omitempty"`
}

type ClaimedRunResult 

type ClaimedRunResult struct {
	Snapshot projectdurable.DurableRunSnapshot `json:"snapshot"`
	ClaimID  string                            `json:"claim_id"`
}

ClaimedRunResult is the metadata-only result of ClaimRunForExecution: the post-claim run snapshot plus the opaque claim token the completion report must echo.

type ExecutionTrace 

type ExecutionTrace struct {
	Input           TestExecutionWorkflowInput             `json:"input"`
	Steps           []projectdurable.DurableActivityResult `json:"steps"`
	FinalStatus     string                                 `json:"final_status"`
	FailureCategory projectdurable.DurableFailureCategory  `json:"failure_category,omitempty"`
}

ExecutionTrace is the Phase 4 workflow result. It mirrors ShadowRunTrace: bounded, safe, metadata-only records of every step plus the mirrored final run status and the first non-none failure category.

type ShadowRunTrace 

type ShadowRunTrace struct {
	Input           AutomationRunWorkflowInput             `json:"input"`
	Steps           []projectdurable.DurableActivityResult `json:"steps"`
	FinalStatus     string                                 `json:"final_status"`
	FailureCategory projectdurable.DurableFailureCategory  `json:"failure_category,omitempty"`
}

ShadowRunTrace is the workflow result: the bounded, safe, metadata-only record of every observation step. FinalStatus mirrors the snapshot's run status; FailureCategory is the first non-none category any step produced.

type StagePlanObservation 

type StagePlanObservation struct {
	Status       string            `json:"status"`
	TaskStatuses map[string]string `json:"task_statuses,omitempty"`
}

type StageTraceEntry 

type StageTraceEntry struct {
	StageRef        string            `json:"stage_ref"`
	WorkflowRef     string            `json:"workflow_ref"`
	PlanID          string            `json:"plan_id,omitempty"`
	TaskCount       int               `json:"task_count"`
	ReviewerCount   int               `json:"reviewer_count"`
	AutomationCount int               `json:"automation_count"`
	Status          string            `json:"status,omitempty"`
	TaskStatuses    map[string]string `json:"task_statuses,omitempty"`
	CarriedIDsCount int               `json:"carried_ids_count"`
}

type TestExecutionWorkflowInput 

type TestExecutionWorkflowInput struct {
	ProjectID    string                               `json:"project_id"`
	AutomationID string                               `json:"automation_id"`
	RunID        string                               `json:"run_id"`
	TaskID       string                               `json:"task_id,omitempty"`
	TraceID      string                               `json:"trace_id,omitempty"`
	RunnerID     string                               `json:"runner_id"`
	Outcome      projectdurable.DurableAttemptOutcome `json:"outcome"`
}

TestExecutionWorkflowInput is the metadata-only input of the Phase 4 TEST-ONLY execution workflow. No production path constructs it: there is no configuration that enables durable execution, and the only construction sites are tests. Outcome carries the attempt completion the simulated runner will report; the workflow injects the claim token after claiming.

type WorkflowChainActivities 

type WorkflowChainActivities struct {
	Pipeline projectdurable.ChainPipelinePort
	Compare  projectdurable.ChainRunComparator
	Shadow   projectdurable.ShadowComparisonWriter
}

func (*WorkflowChainActivities) ActivateStagePlan 

func (a *WorkflowChainActivities) ActivateStagePlan(ctx context.Context, planID string) (projectdurable.DurableActivityResult, error)

func (*WorkflowChainActivities) CarryForwardOutputs 

func (a *WorkflowChainActivities) CarryForwardOutputs(ctx context.Context, fromPlanID, toStageRef string) (CarriedTaskIDs, error)

func (*WorkflowChainActivities) CompileStage 

func (a *WorkflowChainActivities) CompileStage(ctx context.Context, projectID, chainRef string, stage projectdurable.SafeChainStagePlan, inputRef string, carriedTaskIDs []string) (projectdurable.SafeStageCompileOutcome, error)

func (*WorkflowChainActivities) ObserveGitOps 

func (a *WorkflowChainActivities) ObserveGitOps(ctx context.Context, chainRef string) (ChainGitOpsObservation, error)

func (*WorkflowChainActivities) ObserveStagePlan 

func (a *WorkflowChainActivities) ObserveStagePlan(ctx context.Context, planID string) (StagePlanObservation, error)

func (*WorkflowChainActivities) ReleaseCompiledTasks 

func (a *WorkflowChainActivities) ReleaseCompiledTasks(ctx context.Context, planID string) (projectdurable.DurableActivityResult, error)

func (*WorkflowChainActivities) ResolveChainStages 

func (a *WorkflowChainActivities) ResolveChainStages(ctx context.Context, projectID, chainRef string) ([]projectdurable.SafeChainStagePlan, error)

func (*WorkflowChainActivities) WriteChainShadowComparison 

func (a *WorkflowChainActivities) WriteChainShadowComparison(ctx context.Context, input ChainShadowWorkflowInput, trace ChainShadowTrace) (projectdurable.DurableActivityResult, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.