converge

README

Converge

Converge is an embeddable Go reconciliation engine for Agent runtimes. It accepts complete desired snapshots, builds generation-aware plans, executes provider operations, and persists enough state to recover conservatively after a process restart.

The module exposes one recommended entry package and three supporting packages:

• converge: recommended Agent integration boundary, including durable snapshot acceptance and status APIs;
• core: reconciliation, plans, attempts, external effects, recovery, and SQLite/in-memory storage implementations;
• observability: backend-neutral tracing and transition contracts plus a bounded asynchronous adapter;
• pkg/model: wire and lifecycle model types.

Embedding

ctx, cancel := context.WithCancel(context.Background())

runtime, err := converge.OpenSQLiteRuntime(
    ctx,
    "/var/lib/my-agent/converge.db",
    core.WithLogger(logger),
    core.WithObserver(observer),
)
if err != nil {
    return err
}

if err := runtime.RegisterProvider(ctx, provider); err != nil {
	_ = runtime.Close()
    return err
}

runErr := runtime.Run(ctx)
// Runtime.Run returns only after tracked planning, execution, control,
// deletion, verification, and outbox workers have stopped.
if err := runtime.Close(); err != nil {
    return err
}
return runErr

For normal shutdown, call cancel and wait for Runtime.Run to return before closing the SQLite store or Provider resources. Providers must honor context cancellation; Go cannot forcibly stop an implementation that ignores it. Runtime.Run and Reconciler.Run are single-use and reject a second invocation.

OpenSQLiteRuntime owns its SQLite store. After Run returns, call Runtime.Close to release it. The caller still owns Provider resources and any observability.AsyncObserver.Shutdown.

Advanced users that need custom storage or event-bus wiring may use the core package directly and assume its lifecycle invariants.

Provider lifecycle

Register Providers before starting the runtime. A plan remains bound to the Provider digest that created it, so rolling upgrades retain older Provider instances while durable work may still reference them. When the runtime is stopped, UnregisterProviderVersion safely removes an unused, non-current version and rejects removal of a version still referenced by durable state.

Durability and protocol

See Runtime integration, SQLite durability, External Active Effects, and Observability.

The module currently targets Go 1.26.1. Until a semantic-version release is tagged, consumers should pin an audited commit rather than follow main.

Documentation

Overview

Package converge provides the embeddable durable reconciliation runtime.

Index

• Variables
• type Runtime
  • func OpenSQLiteRuntime(ctx context.Context, path string, options ...core.ReconcilerOption) (*Runtime, error)
  • func (r *Runtime) Backup(ctx context.Context, destination string) error
  • func (r *Runtime) Close() error
  • func (r *Runtime) ConfigStatus(name string) (core.ConfigReport, bool)
  • func (r *Runtime) CurrentSnapshot(ctx context.Context) (SnapshotIdentity, error)
  • func (r *Runtime) Ready() bool
  • func (r *Runtime) Refresh(ctx context.Context, name string) error
  • func (r *Runtime) RegisterProvider(ctx context.Context, provider core.Provider) error
  • func (r *Runtime) Run(ctx context.Context) error
  • func (r *Runtime) Status(ctx context.Context) (RuntimeStatus, error)
  • func (r *Runtime) SubmitSnapshot(ctx context.Context, snapshot model.DesiredSnapshot) SnapshotACK
  • func (r *Runtime) UnregisterProviderVersion(providerType, providerDigest string) error
• type RuntimeStatus
• type SnapshotACK
• type SnapshotIdentity

Variables

var (
	ErrRuntimeAlreadyRunning = errors.New("converge runtime Run may only be called once")
	ErrRuntimeRunning        = errors.New("converge runtime is running")
	ErrRuntimeClosed         = errors.New("converge runtime is closed")
)

Types

type Runtime

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

Runtime owns durable snapshot acceptance and translates the latest complete snapshot into idempotent per-config reconciliation intents.

func OpenSQLiteRuntime

func OpenSQLiteRuntime(ctx context.Context, path string, options ...core.ReconcilerOption) (*Runtime, error)

OpenSQLiteRuntime opens a self-contained runtime backed by one SQLite database. Runtime owns the database and closes it from Close.

func (*Runtime) Backup

func (r *Runtime) Backup(ctx context.Context, destination string) error

Backup creates a transactionally consistent copy of the runtime database.

func (*Runtime) Close

func (r *Runtime) Close() error

Close releases the SQLite database owned by a runtime created with OpenSQLiteRuntime. The caller must first cancel Run and wait for it to return. No other Runtime calls may race with Close. Close is idempotent.

func (*Runtime) ConfigStatus

func (r *Runtime) ConfigStatus(name string) (core.ConfigReport, bool)

func (*Runtime) CurrentSnapshot

func (r *Runtime) CurrentSnapshot(ctx context.Context) (SnapshotIdentity, error)

func (*Runtime) Ready

func (r *Runtime) Ready() bool

func (*Runtime) Refresh

func (r *Runtime) Refresh(ctx context.Context, name string) error

func (*Runtime) RegisterProvider

func (r *Runtime) RegisterProvider(ctx context.Context, provider core.Provider) error

RegisterProvider registers a resource implementation. Register Providers before Run; a later digest for the same type is a rolling upgrade.

func (*Runtime) Run

func (r *Runtime) Run(ctx context.Context) error

Run recovers Core first, then replays the latest durable snapshot. A successful SubmitSnapshot does not depend on Run being online.

func (*Runtime) Status

func (r *Runtime) Status(ctx context.Context) (RuntimeStatus, error)

func (*Runtime) SubmitSnapshot

func (r *Runtime) SubmitSnapshot(ctx context.Context, snapshot model.DesiredSnapshot) SnapshotACK

SubmitSnapshot validates and persists the complete snapshot before returning an accepted ACK. Convergence happens asynchronously in Run.

func (*Runtime) UnregisterProviderVersion

func (r *Runtime) UnregisterProviderVersion(providerType, providerDigest string) error

UnregisterProviderVersion removes an unused Provider version while stopped.

type RuntimeStatus

type RuntimeStatus struct {
	Snapshot           SnapshotIdentity    `json:"snapshot"`
	DispatchedRevision uint64              `json:"dispatched_revision,omitempty"`
	ApplyError         string              `json:"apply_error,omitempty"`
	Configs            []core.ConfigReport `json:"configs"`
}

type SnapshotACK

type SnapshotACK struct {
	Accepted bool   `json:"accepted"`
	Revision uint64 `json:"revision"`
	Digest   string `json:"digest"`
	Code     string `json:"code,omitempty"`
	Reason   string `json:"reason,omitempty"`
}

type SnapshotIdentity

type SnapshotIdentity struct {
	Present  bool   `json:"present"`
	Revision uint64 `json:"revision,omitempty"`
	Digest   string `json:"digest,omitempty"`
}