cliworkflow

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: 48 Imported by: 0

Documentation

Overview

Package cliworkflow holds the workflow CLI domain: workflow run, resume, deliver, status, events, approve, reject, cancel, cleanup, delete, and gc commands, the session workflow tool engine, and the workflow snapshot and verifier pinning machinery.

The package must never import internal/cli (the CLI composition root imports this package). Every symbol this package needs from internal/cli is declared below as a nil package variable and assigned by the init() in internal/cli/cliworkflow_wiring.go. Each seam documents the cli helper it stands for and the cycle it breaks. The real fix is to move the helper into a domain package both sides can import (the stack helpers belong in a future internal/clistack; the chat/config helpers belong in the packages that own them).

Index

Constants

View Source 
const WorkflowApprovalDefaultActor = "operator"

WorkflowApprovalDefaultActor names the operator when --actor is not given. It is deterministic (never the ambient OS user) so operator scripts and tests see stable approval records.

View Source 
const WorkflowBranchPrefix = "wf/"

Variables

View Source 
var (
	// ContextStorePath stands for cli.ContextStorePath (context_setup.go).
	ContextStorePath func(root string, cfg config.SubagentConfig) string

	// ApplyPrivacyPolicyFunc stands for cli.applyPrivacyPolicy (chat_command.go).
	ApplyPrivacyPolicyFunc func(res *config.Resolved)

	// LogMCPWarningsFunc stands for cli.logMCPWarnings (limits_summary.go).
	LogMCPWarningsFunc func(w io.Writer, res *config.Resolved)

	// SliceErrorsFunc stands for cli.sliceErrors (errors.go).
	SliceErrorsFunc func(context string, errs []string) error

	// FlagValueFunc stands for cli.flagValue (root.go).
	FlagValueFunc func(args []string, names ...string) (string, []string, bool, error)

	// FlagVarFunc stands for cli.flagVar (root.go).
	FlagVarFunc func(args []string, names ...string) ([]string, []string, bool, error)

	// InstallHookSessionFunc stands for cli.installHookSession (hooks_command.go).
	InstallHookSessionFunc func(workspaceRoot string, staleBypass, quiet bool) (func(), error)

	// LoadChatSkillsFunc stands for cli.loadChatSkills (chat_command.go).
	LoadChatSkillsFunc func(wsRoot string) (*skills.Registry, error)

	// NewSessionDispatcherFunc stands for cli.NewSessionDispatcher (dispatcher.go).
	NewSessionDispatcherFunc func(opts cliagents.SessionDispatcherOpts) (*runtime.Dispatcher, error)

	// InitCoordinatorFunc stands for cli.initCoordinator (orchestration_state.go).
	InitCoordinatorFunc func(d *runtime.Dispatcher, cfg config.SubagentConfig, repos ...ledger.LedgerRepository) coordinator.Coordinator

	// InjectBaselineMessagingFunc stands for cli.injectBaselineMessaging
	// (messaging_tools.go).
	InjectBaselineMessagingFunc func(full, scoped *tools.Registry, cfg config.SubagentConfig, disallowed map[string]struct{})

	// MessagingDisallowedFunc stands for cli.messagingDisallowed
	// (agent_task_handler.go).
	MessagingDisallowedFunc func(names []string) map[string]struct{}

	// SessionAutoDeliveryRepairLoopFunc stands for
	// cli.sessionAutoDeliveryRepairLoop (session_delivery_repair.go).
	SessionAutoDeliveryRepairLoopFunc func(runCtx context.Context, repo workflowledger.Repository, root string, res *config.Resolved, store *storage.SQLite, runID string, advance func(context.Context) (workflowledger.RunSnapshot, error), driveStack func(context.Context) (bool, error), deliverPlanRun bool)

	// ErrStackAwaitsGrant stands for cli.errStackAwaitsGrant
	// (stack_grant_pause.go).
	ErrStackAwaitsGrant error

	// StackingDriveAllowPublishFunc stands for cli.stackingDriveAllowPublish
	// (stack_grant_pause.go).
	StackingDriveAllowPublishFunc func(compiled *definition.CompiledWorkflow) bool

	// ClassifyStackPlanRunDeliveryFunc stands for
	// cli.classifyStackPlanRunDelivery (stack_admit_integration.go).
	ClassifyStackPlanRunDeliveryFunc func(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string, remoteMergeOracle bool) StackPlanRunGate

	// ClassifyStackPlanRunDeliveryFn is the test seam over
	// ClassifyStackPlanRunDeliveryFunc (mirrors the cli var of the same name).
	ClassifyStackPlanRunDeliveryFn func(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string, remoteMergeOracle bool) StackPlanRunGate

	// StackPlanRunFailureReasonFunc stands for cli.stackPlanRunFailureReason
	// (stack_admit_integration.go).
	StackPlanRunFailureReasonFunc func(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string) (failed bool, reason string)

	// StackPlanRunFailureReasonFn is the test seam over
	// StackPlanRunFailureReasonFunc (mirrors the cli var of the same name).
	StackPlanRunFailureReasonFn func(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string) (failed bool, reason string)

	// ErrFailedStackPlanRunFunc stands for cli.errFailedStackPlanRun
	// (stack_admit_integration.go).
	ErrFailedStackPlanRunFunc func(runID, reason string) error

	// ErrUndrivenStackPlanRunFunc stands for cli.errUndrivenStackPlanRun
	// (stack_admit_integration.go).
	ErrUndrivenStackPlanRunFunc func(runID string) error

	// LoadStackPlanOutputFunc stands for cli.loadStackPlanOutput (stack_state.go).
	LoadStackPlanOutputFunc func(repo workflowledger.Repository, stackID string) ([]byte, error)

	// ParseStackPlanOutputFunc stands for cli.parseStackPlanOutput
	// (stack_reconcile.go).
	ParseStackPlanOutputFunc func(raw []byte) (mode string, chunks []delivery.ChunkPlan, hasMore bool, remainingScope string, err error)

	// StackPlanInputsFunc stands for cli.stackPlanInputs (stack_state.go).
	StackPlanInputsFunc func(repo workflowledger.Repository, stackID string) (map[string]string, error)

	// LoadAllStackChunksForDriveFunc stands for
	// cli.loadAllStackChunksForDrive (stack_decompose_continue.go).
	LoadAllStackChunksForDriveFunc func(prepared *PreparedWorkflowRun, stackID string, planOutput []byte, planInputs map[string]string, stdout, stderr io.Writer) (chunks []delivery.ChunkPlan, hasMore bool, hasUnsettledWave bool, remainingScope string, err error)

	// SeedStackLedgerFunc stands for cli.seedStackLedger (stack_state.go).
	SeedStackLedgerFunc func(ledger *workflowledger.Store, stackID string, chunks []delivery.ChunkPlan) error

	// DriveStackToCompletionFunc stands for cli.driveStackToCompletion
	// (stack_drive.go).
	DriveStackToCompletionFunc func(ctx context.Context, prepared *PreparedWorkflowRun, ledger *workflowledger.Store, stackID string, chunks []delivery.ChunkPlan, hasMore bool, hasUnsettledWave bool, remainingScope string, planInputs map[string]string, allowPublish bool, stdout, stderr io.Writer) error

	// LoadAllStackChunksFunc stands for cli.loadAllStackChunks (stack_state.go).
	LoadAllStackChunksFunc func(repo workflowledger.Repository, stackID string) (chunks []delivery.ChunkPlan, hasMore bool, remainingScope string, err error)

	// StackTaskMapFunc stands for cli.stackTaskMap (stack_state.go).
	StackTaskMapFunc func(ledger *workflowledger.Store, stackID string) (map[string]workflowledger.Task, error)

	// StackMergedSetFunc stands for cli.stackMergedSet (stack_state.go).
	StackMergedSetFunc func(byID map[string]workflowledger.Task) map[string]bool

	// AllChunksMergedFunc stands for cli.allChunksMerged (stack_state.go).
	AllChunksMergedFunc func(chunks []delivery.ChunkPlan, merged map[string]bool) bool

	// StackRunRefFunc stands for cli.stackRunRef (stack_reconcile.go).
	StackRunRefFunc func(repo workflowledger.Repository, stackID, chunkID string) (workflowledger.RunSnapshot, bool, error)

	// StackHeadBranchFunc stands for cli.stackHeadBranch (stack_reconcile.go).
	StackHeadBranchFunc func(run workflowledger.RunSnapshot) string

	// StackRunHeadCommitFunc stands for cli.stackRunHeadCommit (stack_state.go).
	StackRunHeadCommitFunc func(repo workflowledger.Repository, run workflowledger.RunSnapshot) string

	// StackRunPushedFunc stands for cli.stackRunPushed (stack_state.go).
	StackRunPushedFunc func(repo workflowledger.Repository, run workflowledger.RunSnapshot) bool

	// StackRunPublishWithheldFunc stands for cli.stackRunPublishWithheld
	// (stack_publish_gate.go).
	StackRunPublishWithheldFunc func(ctx context.Context, repo workflowledger.Repository, runID string, quiet bool) bool

	// StackDecomposedChunksFunc stands for cli.stackDecomposedChunks
	// (stack_admit_integration.go).
	StackDecomposedChunksFunc func(ctx context.Context, repo workflowledger.Repository, runID string) (chunks int, ok bool)

	// OpenContextStoreFunc stands for cli.openContextStore (context_setup.go).
	OpenContextStoreFunc func(root string, cfg config.SubagentConfig) (*storage.SQLite, error)

	// InjectSkillResourceToolFunc stands for cli.InjectSkillResourceTool
	// (skill_resource_tool.go).
	InjectSkillResourceToolFunc func(registry *tools.Registry, activation *skills.SkillActivation) (*tools.Registry, error)

	// GitMergeCheckFunc stands for cli.gitMergeChecker{}.Merged
	// (stack_merge_checker.go).
	GitMergeCheckFunc func(ctx context.Context, git delivery.GitRunner, pr delivery.PRClient, gc delivery.GitContext, headBranch, baseBranch, headCommit, repoSlug string, wasPushed bool) (bool, error)
)
View Source 
var (
	WorkflowBuildLoadSkills    func(wsRoot string) (*skills.Registry, error)
	WorkflowBuildDispatcher    func(opts cliagents.SessionDispatcherOpts) (*runtime.Dispatcher, error)
	WorkflowResumeInstallHooks func(workspaceRoot string, staleBypass, quiet bool) (func(), error)
	WorkflowExecutionHooks     func(workspaceRoot string, staleBypass, quiet bool) (func(), error)
)

Build and resume test seams whose defaults come from cli-owned helpers; InitCLIDefaults fills them from the seams above.

View Source 
var (
	WorkflowDeliverGit   delivery.GitRunner = delivery.RealGit{}
	WorkflowDeliverNewPR                    = func() delivery.PRClient { return delivery.GitHubCLI{} }

	WorkflowDeliveryTimeout = 10 * time.Minute
)
View Source 
var (
	WorkflowResumeOpenStore = OpenWorkflowStore

	WorkflowResumeSetAdmission = func(b WorkflowControllerBuild) error {
		return b.Controller.SetAdmission(b.Admission)
	}
	WorkflowResumeSetForce = func(b WorkflowControllerBuild) error {
		return b.Controller.SetForceResume(true)
	}
	WorkflowResumeRun = func(ctx context.Context, b WorkflowControllerBuild) (workflowledger.RunSnapshot, error) {
		return b.Controller.Run(ctx)
	}
)
View Source 
var (
	WorkflowRunBuild        = buildWorkflowController
	WorkflowRunSetAdmission = func(b WorkflowControllerBuild) error {
		return b.Controller.SetAdmission(b.Admission)
	}
)
View Source 
var SessionDeleteRunFunc = func(ctx context.Context, repo workflowledger.Repository, runID string) error {
	return repo.DeleteRun(ctx, runID)
}

SessionDeleteRunFunc is the test seam over the session engine Delete path's ledger write: it defaults to the repository DeleteRun call so tests can inject a write fault after the operator claim succeeds.

View Source 
var SessionDeliverLedgerReopenFunc = OpenWorkflowReportContext

SessionDeliverLedgerReopenFunc is the test seam over the session engine Deliver path's post-success ledger reopen: it defaults to OpenWorkflowReportContext so tests can inject an open or read fault after a successful delivery.

View Source 
var SettleStackPlanRunIfCompleteFn func(ctx context.Context, prepared *PreparedWorkflowRun, stackID string, stdout io.Writer) error

SettleStackPlanRunIfCompleteFn stands for cli.settleStackPlanRunIfComplete (stack_drive.go): the drive loop's completion settle.

View Source 
var WorkflowAutoDeliveryAttemptTimeout = 30 * time.Minute

WorkflowAutoDeliveryAttemptTimeout bounds one auto-delivery drive+deliver attempt so a stuck stack drive cannot hold the execution flock forever. It mirrors cli.WorkflowAutoDeliveryAttemptTimeout (session_delivery_repair.go).

View Source 
var WorkflowResolutionLockWait = 60 * time.Second

WorkflowResolutionLockWait bounds the execution-lock wait for cancel after stopActive: a settling controller can hold the flock past the cancel wait bound, and a non-blocking acquire would surface as an opaque lock error while the run keeps running. 60s (raised from 5s per docs/architecture/workflow-stack-settle.md's P2): a still-settling deliver (git push + PR + ledger CAS) or a same-run session controller finishing its own auto-delivery repair loop routinely holds the per-run flock for tens of seconds, not five - the short bound surfaced as a false "still held" refusal on legitimate, in-progress work with nothing actually stuck.

It is a var, not a const, ONLY so tests that assert the held-lock refusal can shorten the wait (shortenWorkflowResolutionLockWait). Those tests pin WHICH error a held lock produces, not how long the surface waits, so paying the real wait per surface bought nothing and made internal/cli the critical path of the whole test suite. Production never assigns it.

View Source 
var WorkflowRunsList = func(ctx context.Context, repo workflowledger.Repository, filter ...workflowledger.RunStatus) ([]workflowledger.RunSnapshot, error) {
	return repo.ListRuns(ctx, filter...)
}

WorkflowRunsList is the ledger read seam, overridden in tests to exercise the store-failure path.

View Source 
var WorkflowStackDriveToCompletion func(ctx context.Context, prepared *PreparedWorkflowRun, ledger *workflowledger.Store, stackID string, chunks []delivery.ChunkPlan, hasMore bool, hasUnsettledWave bool, remainingScope string, planInputs map[string]string, allowPublish bool, stdout, stderr io.Writer) error

WorkflowStackDriveToCompletion is the stack driver invoked for a settled multi-chunk plan run. It is a package variable so tests can stub the drive and pin the drive-before-delivery ordering without running chunk agents.

Functions

func AcceptWorkflowSkillChanges 

func AcceptWorkflowSkillChanges(prior *workflowledger.Snapshot, wf *definition.CompiledWorkflow, registry *skills.Registry) ([]string, error)

AcceptWorkflowSkillChanges re-pins drifted skills in the in-memory prior snapshot so verification passes. The durable admission record is never touched: acceptance is per-invocation, mirroring acceptWorkflowVerifierChanges.

func AcquireWorkflowExecutionLock 

func AcquireWorkflowExecutionLock(storePath, runID string) (func(), error)

func ApplyAcceptedSkillChanges 

func ApplyAcceptedSkillChanges(prior *workflowledger.Snapshot, wf *definition.CompiledWorkflow, registry *skills.Registry, stderr io.Writer) error

ApplyAcceptedSkillChanges runs the acceptance rewrite for a resume that passed --accept-skill-change and reports each accepted skill to stderr.

func ApplyWorkflowStoreRoot 

func ApplyWorkflowStoreRoot(res *config.Resolved, root string)

ApplyWorkflowStoreRoot pins workflow execution state to the workspace, regardless of the chat/session store's default. Workflow run IDs and locks are not namespaced across projects the way chat sessions are (see contextWorkspaceID), so letting them fall back to the shared global store would let unrelated workflow runs in different repos collide.

An explicitly set RELATIVE store_path is .mivia namespace notation for the workspace, not process-cwd notation: anchor it to the resolved root too, so every surface that loads config through here opens one store per workspace instead of one per working directory.

A leading ~ expands before relativity is classified: "~/..." names the user-global store shared across workspaces (docs/product/config.md), so it never lands under <root>/~/. Expansion is tilde-only; $VAR text stays literal.

func BeginWorkflowExecution 

func BeginWorkflowExecution(workspaceRoot, storePath, runID string) (func(), error)

func BeginWorkflowExecutionBounded 

func BeginWorkflowExecutionBounded(ctx context.Context, workspaceRoot, storePath, runID string, maxWait time.Duration) (func(), error)

BeginWorkflowExecutionBounded is BeginWorkflowExecution with a bounded wait for a concurrent holder (a settling controller) to release the execution flock: cancel and deliver use it because the plain lock fails with an opaque "lock is busy" error while the in-process controller is still releasing the flock after the cancel wait bound.

func ClaimWorkflowOperator 

func ClaimWorkflowOperator(ctx context.Context, repo workflowledger.Repository, runID, holder string) error

func CleanupWorkflowRunForDialog 

func CleanupWorkflowRunForDialog(runID, root, configPath string) error

CleanupWorkflowRunForDialog runs executeWorkflowCleanup for the workflow-run detail modal, which has nowhere to show the operator-output stream and only needs the error.

func DeliverRunWithStore 

func DeliverRunWithStore(ctx context.Context, root string, res *config.Resolved, store *storage.SQLite, repo workflowledger.Repository, runID string, allowPublish, force bool, stdout, stderr io.Writer) error

DeliverRunWithStore performs delivery for a delivery_pending run. The caller holds the workflow execution file lock (BeginWorkflowExecution).

func DeliveryFaultTransient 

func DeliveryFaultTransient(err error) bool

DeliveryFaultTransient reports whether a delivery failure is a transport fault no agent can repair: a provider-side transient (provider.IsTransient) or a git/gh network death (delivery.IsTransportFault). provider's phrase list is provider-domain and does not know git's texts ("Could not resolve host", "Connection timed out"), which previously misrouted a fetch failure to the repair step and wrote a failed record for it.

func DigestBytes 

func DigestBytes(data []byte) string

func ExecuteWorkflowResume 

func ExecuteWorkflowResume(runID, root, configPath string, force, allowPublish, acceptVerifierChange, acceptSkillChange bool, stdout, stderr io.Writer) error

func ExecuteWorkflowRun 

func ExecuteWorkflowRun(name, root, configPath string, rawInputs []string, allowPublish bool, stdout, stderr io.Writer) error

func HydrateWorkflowSkillSnapshot 

func HydrateWorkflowSkillSnapshot(name string, pinned workflowledger.RefSnapshot) (skills.Definition, []skills.ResourceSnapshot, error)

HydrateWorkflowSkillSnapshot rebuilds the executable skill definition and its resource snapshots from the pinned admission bytes. The pin's digest is re-verified here so a tampered in-memory pin fails closed at dispatch. The hydrated definition carries no source location: resource activation must go through skills.ActivateSnapshot, which serves the pinned bytes from memory.

func InitCLIDefaults 

func InitCLIDefaults()

InitCLIDefaults installs the cli-backed defaults into the package's build and lifecycle seams. internal/cli's wiring init calls it after assigning the seam vars; test mains call it after wiring stubs so a stub wins.

func InstallWorkflowSkillSnapshots 

func InstallWorkflowSkillSnapshots(destination map[string]workflowledger.RefSnapshot, raw []byte) error

func InstallWorkflowSkillSnapshotsFromSnapshot 

func InstallWorkflowSkillSnapshotsFromSnapshot(destination map[string]workflowledger.RefSnapshot, snapshot *workflowledger.Snapshot) error

InstallWorkflowSkillSnapshotsFromSnapshot installs the skill pins the dispatcher enforces at dispatch time directly from a decoded snapshot (the in-memory prior on resume, which may have been re-pinned by --accept-skill-change). Installing from the struct avoids re-marshalling the whole snapshot, so round-trip fidelity of unrelated fields cannot skew the pins, and it guarantees the dispatcher sees exactly the pins verification accepted.

func LoadAgentDefinitionsLocal 

func LoadAgentDefinitionsLocal(workspaceRoot, agentFlag string, skillReg *skills.Registry) (cliagents.AgentLoadResult, error)

LoadAgentDefinitionsLocal loads agent definitions under the user gate. It mirrors cli's test helper of the same name (test_helpers_moved_test.go); duplicated here because Go forbids cross-package _test.go sharing.

func LoadPanelMemberBindings 

func LoadPanelMemberBindings(base string, step definition.Step, registry *agents.AgentRegistry, prior *workflowledger.Snapshot, snapshot *workflowledger.Snapshot) error

func LoadStepReferences 

func LoadStepReferences(base string, step definition.Step, prior *workflowledger.Snapshot) (string, map[string]any, []byte, []byte, error)

func LockWorkflowExecutionFile 

func LockWorkflowExecutionFile(file *os.File) (func(), error)

LockWorkflowExecutionFile wraps the low-level file-lock primitive so the workflow execution lock reports its own name instead of borrowing the Git exclude lock's message. The primitive is not changed; the wording is fixed at this workflow-specific call site. lockWorktreeMarkerFile is shared with the actual Git-exclude marker lock, so every error string it can produce says "Git exclude" - not just the busy case a prior fix special-cased. That left every other failure (permission denied, I/O error, ...) reporting the wrong lock's name to a caller trying to diagnose a stuck workflow delivery or resume. renameGitExcludeLockError rewrites the whole family generically.

func NewCLIWorkflowRunID 

func NewCLIWorkflowRunID() string

func NewSessionWorkflowEngine 

func NewSessionWorkflowEngine(root, configPath string) *sessionWorkflowEngine

NewSessionWorkflowEngine builds the chat-session workflow engine.

func NextStepAfterActive 

func NextStepAfterActive(cw *definition.CompiledWorkflow, activeID string) string

NextStepAfterActive returns the step declared immediately after activeID in the compiled step order, or "" when activeID is not declared or is the last step. The reserved terminals "success" and "failure" are never declared steps, so they cannot be found here.

func OpenContextStorePath 

func OpenContextStorePath(path string) (*storage.SQLite, error)

OpenContextStorePath opens the SQLite context store at path. It mirrors cli.openContextStorePath (context_setup.go); both wrap storage.OpenSQLite.

func OpenWorkflowReportContext 

func OpenWorkflowReportContext(root, configPath string) (workflowledger.Repository, func(), error)

OpenWorkflowReportContext opens the workspace, config, and workflow store for the read-only workflow commands (status, events).

func OpenWorkflowStore 

func OpenWorkflowStore(root string, cfg config.SubagentConfig) (*storage.SQLite, workflowledger.Repository, func(), error)

func PinWorkflowSkills 

func PinWorkflowSkills(raw []byte, wf *definition.CompiledWorkflow, registry *skills.Registry) ([]byte, error)

func ReadWorkflowRef 

func ReadWorkflowRef(base, ref string, max int) ([]byte, error)

ReadWorkflowRef reads one workflow-relative reference (template or schema) with symlink rejection, a size cap, and no path escape.

func RecordAutoDeliveryFailure 

func RecordAutoDeliveryFailure(ctx context.Context, repo workflowledger.Repository, runID string, deliverErr error)

RecordAutoDeliveryFailure durably records an end-of-run auto-delivery failure so the run ledger explains why delivery did not settle: the error text is stored content-addressed and a failed delivery record carrying the ErrorRef is upserted, which `workflow status` and the workflow_status tool surface. In-flight delivery failures already write a failed record via the delivery package; this fills the gap for refusals and pre-flight errors that return before any record write. Best effort: the settled run status (delivery_failed on refusal, delivery_pending on a transient failure) is owned by the delivery path and is never changed here.

func RefuseFailedStackPlanRunDelivery 

func RefuseFailedStackPlanRunDelivery(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string) error

RefuseFailedStackPlanRunDelivery is the `workflow deliver` branch for a plan run whose stack terminally failed. It NEVER publishes: a stack that lost a chunk (or whose integration run reached a status with no outgoing transition) never delivered the plan the PR would describe, so publishing it and settling the run succeeded would report work that does not exist.

It settles as well as refuses, for two reasons. First, the caller holds the run's execution flock (BeginWorkflowExecutionBounded), which is the same precondition the drive and sweep paths settle under, so the CAS cannot race a concurrent driver. Second, refusal alone is a dead end: the failure statuses have no repair edge, so the plan run would sit at delivery_pending with `deliver` refusing it on every retry and no other command able to close it. settleStackPlanRunFailed is a no-op for a run that is absent, already terminal, or not parked at delivery_pending, so a repeated deliver stays idempotent and still returns the refusal for a non-zero exit.

func ResolveWorkflowDialogApproval 

func ResolveWorkflowDialogApproval(runID, approvalID, root, configPath, actor string, reject bool) error

ResolveWorkflowDialogApproval resolves one pending gate approval through the same bounded-lock controller path as the workflow_approval CLI surface. The actor is fixed to WorkflowApprovalDefaultActor so operator scripts and tests see deterministic approval records.

func RunWorkflow 

func RunWorkflow(args []string) error

func RunWorkflowCommandApprove 

func RunWorkflowCommandApprove(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandCancel 

func RunWorkflowCommandCancel(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandCleanup 

func RunWorkflowCommandCleanup(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandDelete 

func RunWorkflowCommandDelete(args []string, workspaceRoot, configPath string, force bool, stdout, stderr io.Writer) error

func RunWorkflowCommandDeliver 

func RunWorkflowCommandDeliver(args []string, workspaceRoot, configPath string, force bool, stdout, stderr io.Writer) error

func RunWorkflowCommandEvents 

func RunWorkflowCommandEvents(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandGC 

func RunWorkflowCommandGC(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

RunWorkflowCommandGC prunes workflow-ledger content rows ("sha256:" prefix) that no live run's events reference any longer - the content a deleted run left behind (mivia workflow delete strips a run's events to a tombstone but never touches the content table). Scoped to the workflow ledger's own ref prefix only: a coordinator/subagent/chat content row (a different prefix in the same table) is never a candidate.

func RunWorkflowCommandReject 

func RunWorkflowCommandReject(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandResume 

func RunWorkflowCommandResume(args []string, workspaceRoot, configPath string, force bool, stdout, stderr io.Writer) error

func RunWorkflowCommandRun 

func RunWorkflowCommandRun(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowCommandRuns 

func RunWorkflowCommandRuns(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

RunWorkflowCommandRuns parses `workflow runs [--status s] [--limit n] [--json]`. It takes no positional arguments.

func RunWorkflowCommandStatus 

func RunWorkflowCommandStatus(args []string, workspaceRoot, configPath string, stdout, stderr io.Writer) error

func RunWorkflowWithIO 

func RunWorkflowWithIO(args []string, stdout, stderr io.Writer) error

func RunWorkflows 

func RunWorkflows(args []string) error

RunWorkflows handles the workflow CLI commands.

func RunWorkflowsWithIO 

func RunWorkflowsWithIO(args []string, stdout, stderr io.Writer) error

func SelectWorkflowWorkspace 

func SelectWorkflowWorkspace(ctx context.Context, sourceRoot, runID string, writeCapable bool, recorded *workflowledger.RunSnapshot) (workflowspace.Identity, func(), error)

func SessionEngineConfigPath 

func SessionEngineConfigPath(root string, res *config.Resolved) string

SessionEngineConfigPath is the config file identity for session workflow tools. Prefer the session Resolved.ConfigPath (covers --config / MIVIA_CONFIG) so read and mutate paths open the same store. Fall back to the workspace project file when no session config is available.

func SettleCLIRunFailure 

func SettleCLIRunFailure(repo workflowledger.Repository, runID string, runErr error)

SettleCLIRunFailure records why a CLI-driven workflow run stopped, for errors the controller did not already settle. Controller.Run self-settles deadline errors (timed_out) and cancel owns cancelled runs, so only genuine non-deadline failures need this settle. It is a no-op for nil, cancelled, and deadline errors.

func SettleFailedStackPlanRunIfNeeded 

func SettleFailedStackPlanRunIfNeeded(ctx context.Context, prepared *PreparedWorkflowRun, runID, cause string) (bool, error)

SettleFailedStackPlanRunIfNeeded fail-settles a delivery_pending stacking plan run whose stack terminally failed (see StackPlanRunFailureReasonFunc), reporting whether it settled. Used by the in-session drive paths so a dead stack settles once instead of being refused as merely incomplete forever.

func SettlePlanRunSkippedDelivery 

func SettlePlanRunSkippedDelivery(ctx context.Context, repo workflowledger.Repository, runID string) error

SettlePlanRunSkippedDelivery settles a plan run whose own publication is disabled (delivery.deliver_plan_run=false) after its stack drove to completion. The run's success terminal parked it at delivery_pending; the stack is done and nothing is published for the run itself, so it CASes to succeeded - the same terminal a delivered run reaches - rather than waiting for a delivery that will never come. It is a no-op when the run no longer waits for delivery (for example a concurrent manual deliver already settled it).

func SettleSessionRunFailure 

func SettleSessionRunFailure(repo workflowledger.Repository, runID string, runErr error)

SettleSessionRunFailure records why a session-driven run stopped, and gives it a terminal status when nothing else will — mirroring the local engine's settleRunFailure, which the session engine previously disagreed with by reading the controller's stop cause only to decide whether to deliver, then dropping it (leaving the run row `running` with no explanation).

Carve-outs: a cancelled run, a run another holder owns, an already-terminal run, and a run parked at delivery_pending/pending/waiting_approval are all left alone — none of them is mid-flight running.

For a run that IS mid-flight, the settle checks the ledger's recovery plan first: the controller persists the completion route before the run-status CAS, so a storage fault between the two can leave a COMPLETED run with a derived terminal ActiveStepID but Status "running". PlanResume detects that and the settle records the plan's real terminal status instead of failing a run that already finished (which would block delivering already-done work). Otherwise the storage fault settles as failed.

func ShortDigest 

func ShortDigest(digest string) string

ShortDigest renders a digest prefix for operator output.

func ShortenWorkflowResolutionLockWaitForTest 

func ShortenWorkflowResolutionLockWaitForTest(t *testing.T)

ShortenWorkflowResolutionLockWaitForTest shortens the resolution lock wait for the calling test and restores it on cleanup.

func SkipParkedPlanRunPublication 

func SkipParkedPlanRunPublication(ctx context.Context, store *storage.SQLite, repo workflowledger.Repository, runID string) bool

SkipParkedPlanRunPublication reports whether a delivery_pending run has the SKIP SHAPE the sweep may settle WITHOUT publishing: the compiled workflow disables the plan run's own publication (delivery.deliver_plan_run false - the default) and the run's task ledger carries the seeded stack plan the chunk drive wrote. The shape predicate alone does NOT authorize the settle: reconcileParkedDelivery settles succeeded only when the stack actually drove to completion (StackDriveCompleted); a seeded-but-incomplete stack stays delivery_pending for the operator to finish with 'mivia stack drive'. Any resolution failure (missing run, corrupt snapshot) returns false, so the run falls through to DeliverRunWithStore, which reports the error exactly as before.

func StackDriveCompleted 

func StackDriveCompleted(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID, policy string, remoteMergeOracle bool) bool

StackDriveCompleted reports whether a stacking plan run's chunk stack actually drove to completion: the run ledger carries the succeeded decompose output the driver reads (LoadStackPlanOutputFunc), every chunk task is merged (AllChunksMergedFunc over StackTaskMapFunc), and the final integration run - keyed by its stable admission key; runID IS the stack id - was admitted and settled. The gate is deliberately STRICTER than waitIntegrationRunSettled (which reports complete for an unsettled integration run), so the sweep never settles a plan run over a stack the driver is still advancing. A delivery_pending integration run counts as settled ONLY under the grant merge policy ("awaits the publish grant"); under merge_policy=auto the driver still auto-merges the integration PR, so delivery_pending is NOT complete there. Any resolution failure (missing run, corrupt output, unseeded plan) returns false: a seeded-but-incomplete stack stays delivery_pending for 'mivia stack drive' to finish.

remoteMergeOracle: settle paths (deliver, stack drive, the sweep) MUST pass true - for a succeeded, pushed integration run under auto policy, the oracle (git merge-base, then gh IsMerged if inconclusive) must confirm the PR actually merged before the plan run may settle. Read-only display surfaces (workflow status's undriven notice, the run_finished event publisher) pass false: the durable pushed evidence settles the DISPLAY verdict, and a read-only surface runs no probes.

func StackPlanMergePolicy 

func StackPlanMergePolicy(ctx context.Context, repo workflowledger.Repository, runID string) string

StackPlanMergePolicy resolves the stacking merge_policy of a plan run from its admitted snapshot (the same snapshot ValidateWorkflowResumeSnapshot validates on the resume path), so StackDriveCompleted can apply the policy-aware delivery_pending rule. Any resolution failure (missing run, corrupt snapshot) returns "" - the grant default: a delivery_pending integration run then counts as complete (admitted, awaiting the publish grant).

func StackPlanRunFailureReasonImpl 

func StackPlanRunFailureReasonImpl(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string) (failed bool, reason string)

StackPlanRunFailureReasonImpl reports whether a multi-chunk stack plan run has already reached a terminal failure state, and why.

func ValidatePanelAgentTools 

func ValidatePanelAgentTools(agent agents.ResolvedAgent, skillName string, opts cliagents.SessionDispatcherOpts, synthesizer bool) error

func ValidateWorkflowResumeSnapshot 

func ValidateWorkflowResumeSnapshot(run workflowledger.RunSnapshot, raw []byte) (workflowledger.Snapshot, *definition.CompiledWorkflow, map[string]any, error)

func VerifyWorkflowSkillSnapshot 

func VerifyWorkflowSkillSnapshot(wf *definition.CompiledWorkflow, registry *skills.Registry, prior *workflowledger.Snapshot) error

VerifyWorkflowSkillSnapshot checks all workflow-referenced skills against the pinned snapshot. It is the unscoped default (nil remaining set); the resume build path calls VerifyWorkflowSkillSnapshotScoped with the PlanResume-derived remaining steps so drift on a step that can never run again does not block the resume (R3).

func VerifyWorkflowSkillSnapshotScoped 

func VerifyWorkflowSkillSnapshotScoped(wf *definition.CompiledWorkflow, registry *skills.Registry, prior *workflowledger.Snapshot, remaining map[string]bool) error

VerifyWorkflowSkillSnapshotScoped checks workflow-referenced skills against the pinned snapshot, scoped to the given set of remaining step IDs. When remaining is nil, all steps are checked (backward-compatible default). When remaining is an empty map, no steps are checked (all completed). When a skill drift is detected the error names the skill, the remaining steps that reference it, and the operator's recovery options (R3+R4).

func WaitForSessionEngineIdle 

func WaitForSessionEngineIdle(t *testing.T, e *sessionWorkflowEngine, runID string)

WaitForSessionEngineIdle blocks until the engine has no active record for runID, failing the test after 5 seconds.

func WaitForSessionEngineIdleWithin 

func WaitForSessionEngineIdleWithin(t *testing.T, e *sessionWorkflowEngine, runID string, within time.Duration)

WaitForSessionEngineIdleWithin blocks until the engine has no active record for runID, failing the test after the given window.

func WireCLIWorkflowProgress 

func WireCLIWorkflowProgress(built *WorkflowControllerBuild, stderr io.Writer)

WireCLIWorkflowProgress attaches the non-interactive JSON progress sink to a freshly built workflow controller. The sink is best-effort: a refusal disables progress reporting instead of failing the run, and a nil controller (resume without a body to re-run) is a no-op.

func WireWorkflowToolOptions 

func WireWorkflowToolOptions(opts *tools.DefaultOptions, root string, res *config.Resolved, provider func() *events.Bus, quiet bool, sessionRepo ledger.LedgerRepository)

WireWorkflowToolOptions attaches Phase 7 workflow tools to DefaultOptions when the workspace has .mivia/workflows/. Reads and mutates share one config identity (session ConfigPath or workspace project file). provider supplies the session event bus for workflow progress lazily, so a bus created after wiring is still observed. nil disables progress publishing.

The parked-delivery sweep (see WorkflowToolServiceWithBus) already runs when provider != nil, so no sweep is launched here. quiet (--quiet) is forwarded to that sweep so the session-start recovery notices honor it.

func WorkflowConfigPath 

func WorkflowConfigPath(root, explicit string) string

func WorkflowRemainingSteps 

func WorkflowRemainingSteps(ctx context.Context, repo workflowledger.Repository, runID string, wf *definition.CompiledWorkflow) (map[string]bool, error)

WorkflowRemainingSteps derives the set of steps a resume may still execute: the PlanResume-derived active step plus every step reachable from it through declared transitions (including partial targets) and on_failure routes. A step outside this set can never run again, so skill drift on it cannot affect the resumed run (R3). When the active step is unknown to the (synthesized) run graph the result is nil: the guard then checks all steps, the fail-closed default.

func WorkflowSkillBytes 

func WorkflowSkillBytes(definition skills.Definition) ([]byte, error)

func WorkflowSkillBytesCurrentAndLegacy 

func WorkflowSkillBytesCurrentAndLegacy(definition skills.Definition) (current, legacy []byte, err error)

WorkflowSkillBytesCurrentAndLegacy marshals the pinned byte shape of one skill in two forms: the current form, whose resource snapshots carry the summary, and the legacy pre-Summary form a binary before the Summary field wrote. Resume verification accepts either shape so a run admitted before the Summary field existed is not reported as drifted after an upgrade (N1); new admissions and re-pins always write the current form.

func WorkflowToolServiceWithBus 

func WorkflowToolServiceWithBus(root string, res *config.Resolved, provider func() *events.Bus, quiet bool, sessionRepo ledger.LedgerRepository) *workflowledger.Service

WorkflowToolServiceWithBus builds the service like workflowToolService and attaches the session event bus provider to its engine. The provider is read at controller attach time, so a bus created after wiring is still observed. provider may be nil - configureChatWorkspace passes nil for a one-shot, non-interactive caller (sessions usage, compact) precisely so the sweep below does not run (F14); only a genuine interactive session (mivia chat) passes a non-nil provider. quiet (--quiet) suppresses the session-start recovery sweep's per-run skip/failure logs, the same way it suppresses the other startup notices.

sessionRepo is the owning chat session's ledger repository - the same value handed to NewSessionDispatcher, so the session's orchestration tools carry the instance the access gate compares. The engine stamps it on every child run it registers. Nil (no session wiring) keeps child-run registration skipped: fail-closed, one notice.

Types

type PreparedWorkflowRun 

type PreparedWorkflowRun struct {
	Root          string
	Res           *config.Resolved
	Store         *storage.SQLite
	Repo          workflowledger.Repository
	CloseFn       func()
	Compiled      *definition.CompiledWorkflow
	Inputs        map[string]any
	InputSnapshot map[string]string
	RefBase       string
	Raw           []byte
}

PreparedWorkflowRun is the immutable input of one workflow run invocation: the opened workspace, store, and compiled definition.

func PrepareWorkflowRun 

func PrepareWorkflowRun(name, root, configPath string, rawInputs []string) (*PreparedWorkflowRun, error)

PrepareWorkflowRun opens the workspace and store and compiles the named workflow with validated inputs, before any execution begins.

type StackPlanRunGate 

type StackPlanRunGate int

StackPlanRunGate classifies a stacking plan run's delivery gate. It mirrors internal/cli's unexported stackPlanRunGate; the wiring converts the cli value with StackPlanRunGate(int(...)).

func ClassifyStackPlanRunDeliveryImpl 

func ClassifyStackPlanRunDeliveryImpl(ctx context.Context, root string, store *storage.SQLite, repo workflowledger.Repository, runID string, remoteMergeOracle bool) StackPlanRunGate

ClassifyStackPlanRunDeliveryImpl reports how runID relates to a multi-chunk stack for delivery purposes: not applicable, incomplete (undriven), complete, or failed. It mirrors the cli implementation.

type WorkflowControllerBuild 

type WorkflowControllerBuild struct {
	Controller *controller.LinearController
	Dispatcher interface{ Close() }
	Admission  controller.Admission
	Cleanup    func()
}

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.