localengine

package
v0.1.3 Latest Latest
Warning

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

Go to latest
Published: Aug 30, 2026 License: AGPL-3.0 Imports: 27 Imported by: 0

Documentation

Overview

Package localengine provides an in-process workflow Engine for agent tools. It reuses controller, ledger, definition, compiler, and delivery packages. Integration tests inject a scripted AgentStepRunner; production hosts may inject a coordinator-backed runner.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NewBusProgressSink

func NewBusProgressSink(bus *events.Bus) controller.ProgressSink

NewBusProgressSink adapts a controller progress sink to an events.Bus: each terminal progress event is published as one events.Event with the workflow kind mapping and run/step attribution, mirroring the session engine's workflowBusProgressSink adapter.

func PanelLimiter

func PanelLimiter() *controller.PanelActorLimiter

PanelLimiter returns the process-wide limiter for workflow panel actors.

func SetProgressSink

func SetProgressSink(s controller.ProgressSink)

SetProgressSink wires the package progress sink. Call it once at startup, before any run, from a single goroutine. A nil sink disables publishing.

Types

type Engine

type Engine struct {
	// WorkspaceRoot is the source workspace for discovery and compilation.
	WorkspaceRoot string
	// Repo is the shared workflow ledger. Required.
	Repo workflowledger.Repository
	// Store is the shared SQLite store backing the stack task ledger
	// (workflowledger.NewStore). Required to drive (or verify the drive of) a
	// multi-chunk stacking plan run; a nil Store degrades the engine to the
	// operator drive (`mivia stack drive`) and refuses delivery of an
	// undriven plan run instead of publishing it.
	Store storage.Store
	// NewRunner builds the agent-step runner for one admitted run.
	// Required for agent steps; a nil NewRunner fails closed (no fake success).
	NewRunner func() controller.AgentStepRunner
	// AgentRegistry supplies immutable agent definitions for panel admission
	// and for agent-step routing pins. Panel work fails closed when the
	// registry cannot resolve a member. Agent steps pin and re-verify their
	// definition digests when the registry is set, and keep the legacy
	// synthetic-digest mode when it is nil.
	AgentRegistry *agents.AgentRegistry
	// NewRunID mints run IDs. Nil uses a secure random wfr- id.
	NewRunID func() string
	// PanelLimiter is the process-wide local actor limiter supplied by the host.
	// A nil value uses the shared workflow process service.
	PanelLimiter *controller.PanelActorLimiter
	// Git and PR are optional delivery adapters.
	Git delivery.GitRunner
	PR  delivery.PRClient
	// DeliveryTimeout bounds one deliver call. Zero uses 2 minutes.
	DeliveryTimeout time.Duration
	// contains filtered or unexported fields
}

Engine runs workflows in-process against a shared ledger repository.

func (*Engine) Cancel

func (e *Engine) Cancel(ctx context.Context, runID string) (workflowledger.CancelResult, error)

Cancel implements workflowledger.Engine.

func (*Engine) Delete

func (e *Engine) Delete(ctx context.Context, runID string, force bool) (workflowledger.DeleteResult, error)

Delete implements workflowledger.Engine. It removes a run from the durable ledger. Settled runs (terminal or delivery_pending) are always deletable; force also permits a non-terminal run (pending/running/waiting_approval) — the crash-recovery override for a run stranded by a dead executor. It mirrors Cancel's fencing: an in-process delivery or controller on this engine refuses, a fresh foreign claim refuses, and only an expired claim may be taken over — deletion must never blind-clear a live delivery claim. The status gate runs BEFORE any claim mutation, so a refused delete leaves claims untouched.

func (*Engine) Deliver

func (e *Engine) Deliver(ctx context.Context, runID string, allowPublish bool) (workflowledger.DeliverResult, error)

Deliver implements workflowledger.Engine.

func (*Engine) EnsureRunWorktreeForTest

func (e *Engine) EnsureRunWorktreeForTest(ctx context.Context, runID string, recorded *workflowledger.RunSnapshot) (Identity, bool)

EnsureRunWorktreeForTest exposes ensureRunWorktree for external tests. Local to the package so coverage can drive the no-workspace-root and recorded-snapshot branches.

func (*Engine) Interrupt

func (e *Engine) Interrupt(runID string) error

Interrupt abandons an in-process controller as if the host process died: open attempts become interrupted, the run stays non-terminal (running), the claim is cleared, and the dying goroutine cannot settle the run.

func (*Engine) ResumeExistingInvocationForTest

func (e *Engine) ResumeExistingInvocationForTest(ctx context.Context, run workflowledger.RunSnapshot, req workflowledger.StartRequest) (workflowledger.StartResult, bool, error)

ResumeExistingInvocationForTest exposes resumeExistingInvocation.

func (*Engine) SetActiveRunForTest

func (e *Engine) SetActiveRunForTest(runID string)

SetActiveRunForTest marks a run as locally active so the resumeExistingInvocation short-circuit triggers.

func (*Engine) Start

Start implements workflowledger.Engine.

func (*Engine) Wait

func (e *Engine) Wait(ctx context.Context, runID string) error

Wait blocks until the background run for runID exits or ctx is done.

type Identity

type Identity struct {
	Root             string
	MainRoot         string
	BaseRef          string
	BaseCommit       string
	OriginBaseCommit string
	WorktreeName     string
	Branch           string
}

Identity records the Git identity of a workflow workspace.

func Ensure

func Ensure(ctx context.Context, sourceRoot, runID string, isolation Isolation) (Identity, error)

Ensure returns the workspace for a new or repeated run admission.

func EnsureRecorded

func EnsureRecorded(ctx context.Context, sourceRoot string, recorded Identity) (Identity, error)

EnsureRecorded recreates a worktree from its immutable admission identity. It never derives a base from the current checkout.

func Resolve

func Resolve(ctx context.Context, sourceRoot string, recorded Identity) (Identity, error)

Resolve validates and returns a recorded workflow workspace.

type Isolation

type Isolation uint8

Isolation selects the workspace policy for a workflow run.

const (
	// IsolationReadOnly uses the caller checkout.
	IsolationReadOnly Isolation = iota
	// IsolationWorktree uses a run-specific Git worktree.
	IsolationWorktree
)

type StaticStepRunner

type StaticStepRunner struct {
	Output     json.RawMessage
	ByStep     map[string]json.RawMessage
	BlockUntil <-chan struct{}
	OnStep     func(controller.AgentStepRequest)
	Err        error
}

StaticStepRunner returns fixed JSON for every agent step (scripted tests). When Err is set, every step fails with that error (used for fail-closed defaults).

func (*StaticStepRunner) RunStep

RunStep implements controller.AgentStepRunner.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL