workflows

package module
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

README

Workflows

github.com/looprig/workflows is an independent Go module for the storage-neutral Flow-to-Harness workflow bridge. Durable storage and concrete transport adapters are supplied by callers; this module does not select a backend.

The bridge is storage-neutral. Callers provide the checkpoint adapter and session-owned services at composition time:

checkpoints := flowstore.New(ledger) // storage.Ledger -> flow.CheckpointStore
registry, _ := workflows.NewRunRegistry(kv)
inputs, _ := workflows.NewInputStore(blobs)
supervisor, _ := workflows.NewSupervisor(workflows.SupervisorConfig{
    SessionID: sessionID, Catalog: catalog, Registry: registry,
    Inputs: inputs, Leaser: leaser,
})

The supervisor owns only bounded run metadata, checkpoint coordination, and metadata-only workflow activities. Application artifacts, policy text, model prompts/output, and report bytes remain in caller-owned stores. Register the supervisor as the Harness session resource so session shutdown cancels its goroutines and releases its lease.

The internal/testworkflow package and bridge integration tests exercise the composition with memory providers, including restart/adoption and activity reconciliation recovery. They are test fixtures, not a production workflow.

Local release provenance

The direct local replacements in go.mod are development inputs, but they are still part of the exact source set used by a standalone Workflows checkpoint. The repository provenance helper discovers those replacements without resolving the network and records one evidence record for Workflows and each replacement.

Each record binds the canonical module path and go.mod SHA-256 to its Git root, HEAD commit, repository tree, module-subdirectory tree, clean state, and a bounded git status --porcelain evidence file. make provenance takes pre- and post-gate captures, verifies their hashes and evidence files, and fails closed if any repository is unavailable, dirty, or drifted. The default status bound is 64 KiB; WORKFLOWS_REPOSITORY_STATUS_MAX_BYTES may lower it but cannot raise the hard 1 MiB ceiling.

The retained WORKFLOWS_PROVENANCE_DIR contains repositories/pre/, repositories/post/, repositories.json, and their status/evidence paths. provenance.json embeds the verified repository record and its SHA-256. Run make repository-provenance-test for deterministic mocked clean, dirty, unavailable, and source/replacement drift coverage.

Offline quality gates

make check is the non-mutating quality gate for this module. It runs fmt-check, vet, the ordinary and race test suites, the bridge integration tests, pinned staticcheck, gosec, govulncheck, a -trimpath build, and module provenance checks. Every gate is explicit; an unavailable tool, module, integration dependency, or vulnerability database is an error rather than a skipped check.

The normal targets set GOWORK=off, GOPROXY=off, GOSUMDB=off, GOTOOLCHAIN=local, and -mod=readonly. Before running make check, provide an already-staged local Go vulnerability database:

POLICY53_GOVULNDB=/absolute/path/to/vulndb-v1 make check

The database must contain the local index/modules.json (or its gzipped form). Network database URLs are rejected. The tagged harness-integration and harness-integration-race composition proofs are required prerequisites of provenance and release-checkpoint; they intentionally use the sibling workspace to supply the test-only inference dependency, while still disabling proxy and checksum resolution. Their successful results are recorded in provenance.json as the required full Harness integration evidence: the selector covers every tagged TestHarness... restore, publisher, cursor-CAS, lease-loss, and fault test.

dependency-check runs go mod verify and resolves the complete module graph offline. dependency-policy additionally requires a reviewed offline license-scanner lock, an exact receipt for the configured regular non-symlinked executable, forbidden-dependency scanning, and a complete license report. The lock, rather than the receipt, pins the immutable source revision/source digest, distribution kind/digest, executable digest, exact invocation and schemas, and canonical receipt identity. The current shared lock is explicitly unsupported until real scanner inputs are reviewed, so a local executable cannot make the gate pass. For a reviewed archive lock, WORKFLOWS_LICENSE_SCANNER_DISTRIBUTION identifies the exact local artifact; standalone executable locks bind the distribution digest to the configured bytes. Signature metadata is recorded as unsupported because no repository signature convention or verifier is available.

The receipt must match those lock-pinned values, the canonical executable path, lock SHA-256, offline network declaration, and exact invocation. The report must repeat the same scanner identity, source and distribution/executable digests, canonical receipt identity, lock digest, invocation, and the SHA-256 identity of the exact receipt file before its module/license set is accepted; a merely nonempty approved report is insufficient. notices-check verifies the committed notice boundary. provenance retains two compared offline module inventories, their SHA-256 digests, the go mod verify result, and a structured provenance.json record containing the deterministic Go commands, toolchain, all tagged Harness integration results (ordinary and race), and any explicitly supplied build command/flags. Set WORKFLOWS_PROVENANCE_DIR to an absolute private output directory to retain the evidence, or set WORKFLOWS_PROVENANCE_EXPECTED to compare it with a previously reviewed inventory. This module does not own Policy53’s NIST, AnyDoc, or render locks. In the workspace, point WORKFLOWS_LICENSE_SCANNER_LOCK at the reviewed scanner lock (or provide an equivalent reviewed lock explicitly); missing scanner inputs fail closed.

For the release/checkpoint path, run make release-checkpoint. Its wrapper captures repository evidence before the ordinary format, test, race, integration, tool, and build gates; provenance then runs the dependency, notice, and both complete tagged Harness integration suites before taking the post-capture. The checkpoint cannot report success if any source or replacement repository is unavailable, dirty, or drifted. Both tagged Harness commands are recorded as PASS only after their normal and race tests have completed successfully. The ordinary harness-integration target remains available for the bounded non-race run; both tagged targets select ^TestHarness and retain their 90-second and 180-second timeouts.

Documentation

Overview

Package workflows contains the storage-neutral workflow bridge and its typed workflow definitions.

Index

Constants

View Source 
const (
	MaxRunRecordBytes      = 64 << 10
	MaxArtifactReferences  = 128
	MaxRunPageSize         = 100
	DefaultRunPageSize     = 50
	MaxInputBytes          = MaxDocumentBytes
	ArtifactInputBootstrap = "bootstrap"
	ArtifactInputParent    = "parent"
)
View Source 
const (
	MaxSchemaBytes        = 64 << 10
	MaxSchemaDepth        = 32
	MaxSchemaProperties   = 1024
	MaxDocumentBytes      = 256 << 10
	MaxDocumentDepth      = 32
	MaxDocumentProperties = 4096
)
View Source 
const (
	MaxStatusSummaryBytes = 1024
)
View Source 
const SupervisorResourceName = "policy53-workflow-supervisor"

Variables

View Source 
var (
	ErrActivityValidation = errors.New("workflow activity validation failed")
	ErrReconciliation     = errors.New("workflow activity reconciliation failed")
)
View Source 
var (
	ErrUnknownDefinition   = errors.New("unknown workflow definition")
	ErrDuplicateDefinition = errors.New("duplicate workflow definition")
	ErrInvalidSchema       = errors.New("invalid workflow schema")
	ErrInvalidInput        = errors.New("invalid workflow input")
)
View Source 
var (
	ErrNotFound      = errors.New("workflow record not found")
	ErrConflict      = errors.New("workflow record conflict")
	ErrCorruptRecord = errors.New("corrupt workflow record")
)
View Source 
var (
	ErrSupervisorActive = errors.New("workflow supervisor already active")
	ErrSupervisorClosed = errors.New("workflow supervisor closed")
	ErrSessionOwned     = errors.New("workflow session already owned")
	ErrAdoption         = errors.New("workflow adoption failed")
	ErrShutdownTimeout  = errors.New("workflow supervisor shutdown timed out")
)

Functions

func StrictJSONDecoder 

func StrictJSONDecoder[S any](raw json.RawMessage) (S, error)

Types

type ActivityHistoryPage 

type ActivityHistoryPage struct {
	Records      []ActivityHistoryRecord
	NextRevision *uint64
	NextEventID  *uuid.UUID
}

ActivityHistoryPage is a bounded page of projected workflow activities. NextRevision is the exclusive revision cursor for the next request; a zero value means there is no next page.

type ActivityHistoryRecord 

type ActivityHistoryRecord struct {
	Revision uint64
	Metadata tool.WorkflowActivityMetadata
}

ActivityHistoryRecord is the safe, projected view of one durable workflow activity. It intentionally contains Harness metadata only; Flow state, checkpoint payloads, policy text, and model output never cross this API.

type ActivityValidationError 

type ActivityValidationError struct {
	Field string
	Rule  string
}

func (*ActivityValidationError) Error 

func (e *ActivityValidationError) Error() string

func (*ActivityValidationError) Unwrap 

func (e *ActivityValidationError) Unwrap() error

type AdoptionError 

type AdoptionError struct {
	RunID uuid.UUID
	Op    string
	Err   error
}

func (*AdoptionError) Error 

func (e *AdoptionError) Error() string

func (*AdoptionError) Unwrap 

func (e *AdoptionError) Unwrap() error

type ArtifactReference 

type ArtifactReference struct {
	ID     string `json:"id"`
	Kind   string `json:"kind"`
	Digest string `json:"digest"`
	Size   int64  `json:"size"`
}

ArtifactReference identifies immutable output or intermediate artifact bytes.

type Catalog 

type Catalog struct {
	// contains filtered or unexported fields
}

func NewCatalog 

func NewCatalog() *Catalog

func (*Catalog) List 

func (c *Catalog) List() []Metadata

func (*Catalog) Register 

func (c *Catalog) Register(definition Definition) error

func (*Catalog) Resolve 

func (c *Catalog) Resolve(name, version string) (Definition, error)

type ConflictError 

type ConflictError struct {
	SessionID uuid.UUID
	RunID     uuid.UUID
	Expected  uint64
	Actual    uint64
	Reason    string
}

ConflictError reports create/CAS conflicts and rejected lifecycle mutations.

func (*ConflictError) Error 

func (e *ConflictError) Error() string

func (*ConflictError) Unwrap 

func (e *ConflictError) Unwrap() error

type CorruptRecordError 

type CorruptRecordError struct {
	Key string
	Err error
}

CorruptRecordError reports malformed or inconsistent durable metadata.

func (*CorruptRecordError) Error 

func (e *CorruptRecordError) Error() string

func (*CorruptRecordError) Unwrap 

func (e *CorruptRecordError) Unwrap() error

type Definition 

type Definition interface {
	Metadata() Metadata
	ValidateInput(json.RawMessage) (ValidatedInput, error)
	ValidateResume(json.RawMessage) (ValidatedResume, error)
	Start(context.Context, ValidatedInput, ...flow.RunOption) (*Result, error)
	Resume(context.Context, flow.GraphRunID, ValidatedResume, ...flow.RunOption) (*Result, error)
	Get(context.Context, flow.GraphRunID) (*Result, error)
	History(context.Context, flow.GraphRunID) ([]flow.GraphRunState, error)
	Cancel(context.Context, flow.GraphRunID, string, ...flow.RunOption) error
	// contains filtered or unexported methods
}

type DuplicateDefinitionError 

type DuplicateDefinitionError struct{ Name, Version string }

func (*DuplicateDefinitionError) Error 

func (e *DuplicateDefinitionError) Error() string

func (*DuplicateDefinitionError) Unwrap 

func (e *DuplicateDefinitionError) Unwrap() error

type InputReference 

type InputReference struct {
	Digest string `json:"digest"`
	Key    string `json:"key"`
	Size   int64  `json:"size"`
}

InputReference identifies schema-validated input bytes held outside KV.

type InputStore 

type InputStore struct {
	// contains filtered or unexported fields
}

InputStore persists validated JSON inputs in session-private immutable blobs.

func NewInputStore 

func NewInputStore(blobs storage.Blobs) (*InputStore, error)

func (*InputStore) Get 

func (s *InputStore) Get(ctx context.Context, sessionID uuid.UUID, ref InputReference) ([]byte, error)

func (*InputStore) Put 

func (s *InputStore) Put(ctx context.Context, sessionID uuid.UUID, canonicalJSON []byte) (InputReference, error)

type InvalidInputError 

type InvalidInputError struct {
	Field string
	Err   error
}

func (*InvalidInputError) Error 

func (e *InvalidInputError) Error() string

func (*InvalidInputError) Unwrap 

func (e *InvalidInputError) Unwrap() error

type InvalidSchemaError 

type InvalidSchemaError struct {
	Field string
	Err   error
}

func (*InvalidSchemaError) Error 

func (e *InvalidSchemaError) Error() string

func (*InvalidSchemaError) Unwrap 

func (e *InvalidSchemaError) Unwrap() error

type ListRunsRequest 

type ListRunsRequest struct {
	After string
	Limit int
}

type Metadata 

type Metadata struct {
	// contains filtered or unexported fields
}

func NewMetadata 

func NewMetadata(name, version, description string, inputSchema, resumeSchema json.RawMessage, vertices []VertexMetadata) (Metadata, error)

func (Metadata) Description 

func (m Metadata) Description() string

func (Metadata) InputSchema 

func (m Metadata) InputSchema() json.RawMessage

func (Metadata) MarshalJSON 

func (m Metadata) MarshalJSON() ([]byte, error)

func (Metadata) Name 

func (m Metadata) Name() string

func (Metadata) ResumeSchema 

func (m Metadata) ResumeSchema() json.RawMessage

func (Metadata) Version 

func (m Metadata) Version() string

func (Metadata) Vertices 

func (m Metadata) Vertices() []VertexMetadata

type NotFoundError 

type NotFoundError struct {
	Kind      string
	SessionID uuid.UUID
	RunID     uuid.UUID
	Digest    string
}

NotFoundError reports a session-scoped run or private input that is absent.

func (*NotFoundError) Error 

func (e *NotFoundError) Error() string

func (*NotFoundError) Unwrap 

func (e *NotFoundError) Unwrap() error

type ReconciliationError 

type ReconciliationError struct {
	RunID uuid.UUID
	Op    string
	Err   error
}

func (*ReconciliationError) Error 

func (e *ReconciliationError) Error() string

func (*ReconciliationError) Unwrap 

func (e *ReconciliationError) Unwrap() error

type Result 

type Result struct {
	Run        flow.GraphRunState
	State      json.RawMessage
	Interrupts []flow.Interruption
	Halt       *flow.Halt
	Summary    string
}

type ResumeDecoder 

type ResumeDecoder func(json.RawMessage) (any, error)

func TypedResumeDecoder 

func TypedResumeDecoder[R any](decoder StateDecoder[R]) ResumeDecoder

type Run 

type Run struct {
	SessionID         uuid.UUID       `json:"session_id"`
	ToolExecutionID   uuid.UUID       `json:"tool_execution_id"`
	DefinitionName    string          `json:"definition_name"`
	DefinitionVersion string          `json:"definition_version"`
	ID                uuid.UUID       `json:"id"`
	GraphRunID        flow.GraphRunID `json:"graph_run_id"`
	ParentRunID       uuid.UUID       `json:"parent_run_id,omitzero"`
	// ArtifactSessionID/ArtifactRunID identify the private namespace owned by
	// this workflow execution. They are deliberately separate from SessionID
	// and ID, which identify the Harness/workflow record.
	ArtifactSessionID      uuid.UUID           `json:"artifact_session_id"`
	ArtifactRunID          uuid.UUID           `json:"artifact_run_id"`
	ArtifactInputKind      string              `json:"artifact_input_kind,omitempty"`
	ArtifactInputSessionID uuid.UUID           `json:"artifact_input_session_id,omitzero"`
	ArtifactInputRunID     uuid.UUID           `json:"artifact_input_run_id,omitzero"`
	Input                  InputReference      `json:"input"`
	Status                 RunStatus           `json:"status"`
	StatusSummary          string              `json:"status_summary,omitempty"`
	CancelRequested        bool                `json:"cancel_requested,omitempty"`
	CheckpointRevision     uint64              `json:"checkpoint_revision"`
	ActivityCursor         uint64              `json:"activity_cursor"`
	LedgerLocator          string              `json:"ledger_locator"`
	Artifacts              []ArtifactReference `json:"artifacts,omitempty"`
	CreatedAt              time.Time           `json:"created_at"`
	UpdatedAt              time.Time           `json:"updated_at"`
	Revision               uint64              `json:"-"`
}

Run is the bounded durable metadata for one session-owned workflow execution. Revision is the storage.KV CAS token and is not encoded into the value.

type RunPage 

type RunPage struct {
	Runs []Run
	Next string
}

type RunRegistry 

type RunRegistry struct {
	// contains filtered or unexported fields
}

RunRegistry persists bounded session-owned workflow metadata in neutral KV.

func NewRunRegistry 

func NewRunRegistry(kv storage.KV) (*RunRegistry, error)

func (*RunRegistry) CompareAndSwap 

func (r *RunRegistry) CompareAndSwap(ctx context.Context, expectedRevision uint64, next Run) (*Run, error)

func (*RunRegistry) Create 

func (r *RunRegistry) Create(ctx context.Context, run Run) (*Run, error)

func (*RunRegistry) Get 

func (r *RunRegistry) Get(ctx context.Context, sessionID, runID uuid.UUID) (*Run, error)

func (*RunRegistry) List 

func (r *RunRegistry) List(ctx context.Context, sessionID uuid.UUID, request ListRunsRequest) (RunPage, error)

type RunStatus 

type RunStatus string

RunStatus is the durable lifecycle state of a registered workflow run. 

const (
	RunPending     RunStatus = "pending"
	RunRunning     RunStatus = "running"
	RunInterrupted RunStatus = "interrupted"
	RunCompleted   RunStatus = "completed"
	RunCancelled   RunStatus = "cancelled"
	RunFailed      RunStatus = "failed"
)

type SessionOwnedError 

type SessionOwnedError struct {
	SessionID   uuid.UUID
	HolderEpoch uint64
}

func (*SessionOwnedError) Error 

func (e *SessionOwnedError) Error() string

func (*SessionOwnedError) Unwrap 

func (e *SessionOwnedError) Unwrap() error

type StateDecoder 

type StateDecoder[S any] func(json.RawMessage) (S, error)

type StatusSummarizer 

type StatusSummarizer[S any] func(S) string

type Supervisor 

type Supervisor struct {
	// contains filtered or unexported fields
}

func NewSupervisor 

func NewSupervisor(config SupervisorConfig) (*Supervisor, error)

func (*Supervisor) Activate 

func (s *Supervisor) Activate(ctx context.Context, services tool.SessionResourceServices) error

func (*Supervisor) Cancel 

func (s *Supervisor) Cancel(ctx context.Context, runID uuid.UUID, reason string) error

func (*Supervisor) History 

func (s *Supervisor) History(ctx context.Context, runID uuid.UUID, afterRevision uint64, afterEventID uuid.UUID, limit int) (ActivityHistoryPage, error)

History returns the projected durable activity history for one session run. afterRevision is the first revision to consider. A nonzero afterEventID resumes after that specific activity when a single checkpoint revision was too large for one page. Reads are serialized with the run controller so a concurrent lifecycle transition cannot produce a mixed registry/checkpoint view.

func (*Supervisor) LastError 

func (s *Supervisor) LastError() error

func (*Supervisor) Resume 

func (s *Supervisor) Resume(ctx context.Context, runID uuid.UUID, payload json.RawMessage) error

func (*Supervisor) SessionID 

func (s *Supervisor) SessionID() uuid.UUID

SessionID returns the immutable Harness session owner for this supervisor. The value is exposed for process-resource composition to reject a resource accidentally created for a different session before it reaches a tool.

func (*Supervisor) Shutdown 

func (s *Supervisor) Shutdown(ctx context.Context) error

func (*Supervisor) Start 

func (s *Supervisor) Start(ctx context.Context, runID uuid.UUID) (<-chan struct{}, <-chan error, error)

Start schedules one pending run on this session-owned supervisor and returns channels for the durable revision-zero seed acknowledgement and any definite scheduling/start failure. The worker continues through workflow completion; callers only wait on seeded when they need the non-blocking start contract.

func (*Supervisor) WaitIdle 

func (s *Supervisor) WaitIdle(ctx context.Context) error

type SupervisorConfig 

type SupervisorConfig struct {
	SessionID       uuid.UUID
	Catalog         *Catalog
	Registry        *RunRegistry
	Inputs          *InputStore
	Leaser          storage.Leaser
	Now             func() time.Time
	ShutdownTimeout time.Duration
	MaxWorkers      int
}

type TypedDefinition 

type TypedDefinition[S any] struct {
	// contains filtered or unexported fields
}

func NewTypedDefinition 

func NewTypedDefinition[S any](metadata Metadata, runner *flow.Runner[S], store flow.CheckpointStore, stateDecoder StateDecoder[S], resumeDecoder ResumeDecoder, summarizer StatusSummarizer[S]) (*TypedDefinition[S], error)

func (*TypedDefinition[S]) Adopt 

func (d *TypedDefinition[S]) Adopt(ctx context.Context, id flow.GraphRunID, opts ...flow.RunOption) (*Result, error)

Adopt continues a durable running Flow checkpoint after a supervisor restart. It is intentionally separate from Resume: adoption carries no user payload and is only selected after the supervisor observes a running, nonterminal checkpoint. Interrupted checkpoints still require ValidateResume and an explicit user action.

func (*TypedDefinition[S]) Cancel 

func (d *TypedDefinition[S]) Cancel(ctx context.Context, id flow.GraphRunID, reason string, opts ...flow.RunOption) error

func (*TypedDefinition[S]) Get 

func (d *TypedDefinition[S]) Get(ctx context.Context, id flow.GraphRunID) (*Result, error)

func (*TypedDefinition[S]) History 

func (d *TypedDefinition[S]) History(ctx context.Context, id flow.GraphRunID) ([]flow.GraphRunState, error)

func (*TypedDefinition[S]) Metadata 

func (d *TypedDefinition[S]) Metadata() Metadata

func (*TypedDefinition[S]) Resume 

func (d *TypedDefinition[S]) Resume(ctx context.Context, id flow.GraphRunID, resume ValidatedResume, opts ...flow.RunOption) (*Result, error)

func (*TypedDefinition[S]) Start 

func (d *TypedDefinition[S]) Start(ctx context.Context, input ValidatedInput, opts ...flow.RunOption) (*Result, error)

func (*TypedDefinition[S]) ValidateInput 

func (d *TypedDefinition[S]) ValidateInput(raw json.RawMessage) (ValidatedInput, error)

func (*TypedDefinition[S]) ValidateResume 

func (d *TypedDefinition[S]) ValidateResume(raw json.RawMessage) (ValidatedResume, error)

type UnknownDefinitionError 

type UnknownDefinitionError struct{ Name, Version string }

func (*UnknownDefinitionError) Error 

func (e *UnknownDefinitionError) Error() string

func (*UnknownDefinitionError) Unwrap 

func (e *UnknownDefinitionError) Unwrap() error

type ValidatedInput 

type ValidatedInput struct {
	// contains filtered or unexported fields
}

type ValidatedResume 

type ValidatedResume struct {
	// contains filtered or unexported fields
}

type VertexMetadata 

type VertexMetadata struct {
	// contains filtered or unexported fields
}

func NewVertexMetadata 

func NewVertexMetadata(label string) VertexMetadata

func NewVertexMetadataForID 

func NewVertexMetadataForID(id flow.VertexID, label string) VertexMetadata

NewVertexMetadataForID binds a safe display label to Flow's stable vertex identity. Definitions that provide IDs let activity projection remain correct when the graph completes vertices out of declaration order.

func (VertexMetadata) ID 

func (m VertexMetadata) ID() flow.VertexID

func (VertexMetadata) Label 

func (m VertexMetadata) Label() string

func (VertexMetadata) MarshalJSON 

func (m VertexMetadata) MarshalJSON() ([]byte, error)

Source Files

View all Source files

Directories

Path Synopsis
examples
docs/stage18_workflows command
internal
testworkflow
Package testworkflow contains a small deterministic graph used only by the bridge integration and recovery tests.
 Package testworkflow contains a small deterministic graph used only by the bridge integration and recovery tests.

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.