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 ¶
- func NewBusProgressSink(bus *events.Bus) controller.ProgressSink
- func PanelLimiter() *controller.PanelActorLimiter
- func SetProgressSink(s controller.ProgressSink)
- type Engine
- func (e *Engine) Cancel(ctx context.Context, runID string) (workflowledger.CancelResult, error)
- func (e *Engine) Delete(ctx context.Context, runID string, force bool) (workflowledger.DeleteResult, error)
- func (e *Engine) Deliver(ctx context.Context, runID string, allowPublish bool) (workflowledger.DeliverResult, error)
- func (e *Engine) EnsureRunWorktreeForTest(ctx context.Context, runID string, recorded *workflowledger.RunSnapshot) (Identity, bool)
- func (e *Engine) Interrupt(runID string) error
- func (e *Engine) ResumeExistingInvocationForTest(ctx context.Context, run workflowledger.RunSnapshot, ...) (workflowledger.StartResult, bool, error)
- func (e *Engine) SetActiveRunForTest(runID string)
- func (e *Engine) Start(ctx context.Context, req workflowledger.StartRequest) (workflowledger.StartResult, error)
- func (e *Engine) Wait(ctx context.Context, runID string) error
- type Identity
- type Isolation
- type StaticStepRunner
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 ¶
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 ¶
SetActiveRunForTest marks a run as locally active so the resumeExistingInvocation short-circuit triggers.
func (*Engine) Start ¶
func (e *Engine) Start(ctx context.Context, req workflowledger.StartRequest) (workflowledger.StartResult, error)
Start implements workflowledger.Engine.
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 EnsureRecorded ¶
EnsureRecorded recreates a worktree from its immutable admission identity. It never derives a base from the current checkout.
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 ¶
func (r *StaticStepRunner) RunStep(ctx context.Context, req controller.AgentStepRequest) (controller.AgentStepResult, error)
RunStep implements controller.AgentStepRunner.
Source Files
¶
- engine.go
- engine_admission.go
- engine_cancel.go
- engine_delete.go
- engine_deliver.go
- engine_invocation.go
- engine_progress.go
- engine_resume.go
- engine_resume_worktree.go
- engine_run_identity.go
- engine_stack.go
- engine_stack_settle.go
- engine_start.go
- fence.go
- helpers.go
- interrupt.go
- panel_limiter.go
- resume_admission.go
- runner.go
- runs_summary.go
- test_helpers_export.go
- workspace.go
- worktree.go