flowctx.go

v0.6.1

package legatus

import (
	"context"
	"io"
	"log/slog"
	"sync"
)

// FlowCtx threads through the steps of one flow execution on one host. Its param
// bag is private to this execution, so steps pass data forward without locking
// (IV3). Cross-host data goes through Shared.
type FlowCtx struct {
	ctx      context.Context
	host     *Host
	conn     Conn
	log      *slog.Logger
	dryRun   bool
	prompter Prompter
	shared   *SharedStore
	params   map[string]any
	changed  map[string]bool
	out      io.Writer
}

// FlowCtxOption configures a FlowCtx.
type FlowCtxOption func(*FlowCtx)

// WithFlowLogger sets the per-flow logger.
func WithFlowLogger(l *slog.Logger) FlowCtxOption {
	return func(fc *FlowCtx) {
		if l != nil {
			fc.log = l
		}
	}
}

// WithFlowDryRun toggles dry-run.
func WithFlowDryRun(d bool) FlowCtxOption { return func(fc *FlowCtx) { fc.dryRun = d } }

// WithFlowPrompter sets the prompter.
func WithFlowPrompter(p Prompter) FlowCtxOption {
	return func(fc *FlowCtx) {
		if p != nil {
			fc.prompter = p
		}
	}
}

// WithFlowShared sets the cross-host shared store.
func WithFlowShared(s *SharedStore) FlowCtxOption {
	return func(fc *FlowCtx) {
		if s != nil {
			fc.shared = s
		}
	}
}

// WithFlowOutput sets the human-facing output sink steps stream to (Out()).
func WithFlowOutput(w io.Writer) FlowCtxOption {
	return func(fc *FlowCtx) {
		if w != nil {
			fc.out = w
		}
	}
}

// NewFlowCtx builds a FlowCtx. The Runner uses it; step tests use it directly.
func NewFlowCtx(ctx context.Context, h *Host, c Conn, opts ...FlowCtxOption) *FlowCtx {
	fc := &FlowCtx{
		ctx:      ctx,
		host:     h,
		conn:     c,
		log:      slog.Default(),
		prompter: DenyPrompter{},
		shared:   NewSharedStore(),
		params:   map[string]any{},
		changed:  map[string]bool{},
		out:      io.Discard,
	}
	for _, o := range opts {
		o(fc)
	}
	return fc
}

// Ctx returns the execution context.
func (fc *FlowCtx) Ctx() context.Context { return fc.ctx }

// Host returns the target host.
func (fc *FlowCtx) Host() *Host { return fc.host }

// Conn returns the live connection to the host.
func (fc *FlowCtx) Conn() Conn { return fc.conn }

// Out returns the human-facing output sink. The engine owns it; steps stream
// command output here (never os.Stdout) when asked. Never nil — io.Discard by
// default, so an unconfigured Runner silently discards.
func (fc *FlowCtx) Out() io.Writer { return fc.out }

// Log returns the per-flow logger, pre-tagged by the Runner with host and step.
func (fc *FlowCtx) Log() *slog.Logger { return fc.log }

// DryRun reports whether this is a dry run.
func (fc *FlowCtx) DryRun() bool { return fc.dryRun }

// Prompter returns the injected prompter (non-interactive by default).
func (fc *FlowCtx) Prompter() Prompter { return fc.prompter }

// Shared returns the cross-host store.
func (fc *FlowCtx) Shared() *SharedStore { return fc.shared }

// Set stores a value in the per-flow param bag.
func (fc *FlowCtx) Set(key string, v any) { fc.params[key] = v }

// Get reads a value from the per-flow param bag.
func (fc *FlowCtx) Get(key string) (any, bool) {
	v, ok := fc.params[key]
	return v, ok
}

// setChanged records, by step name, whether a step changed (or would change in
// dry-run). The runner calls it after each step resolves, before the next step's
// Check. Duplicate names OR-accumulate, so two same-named steps collapse to "any
// of them changed".
func (fc *FlowCtx) setChanged(name string, did bool) {
	fc.changed[name] = fc.changed[name] || did
}

// Changed reports whether the named earlier step changed (or would change in
// dry-run) during this flow execution on this host. An unrun or unrecorded step
// reports false.
func (fc *FlowCtx) Changed(name string) bool { return fc.changed[name] }

// AnyChanged reports whether any of the named earlier steps changed. With no
// names, it reports whether any earlier step in this flow changed. Visibility is
// limited to steps already executed on this host (never cross-host).
func (fc *FlowCtx) AnyChanged(names ...string) bool {
	if len(names) == 0 {
		for _, v := range fc.changed {
			if v {
				return true
			}
		}
		return false
	}
	for _, n := range names {
		if fc.changed[n] {
			return true
		}
	}
	return false
}

func (fc *FlowCtx) snapshot() map[string]any {
	out := make(map[string]any, len(fc.params))
	for k, v := range fc.params {
		out[k] = v
	}
	return out
}

// Param reads a typed value from the per-flow param bag.
func Param[T any](fc *FlowCtx, key string) (T, bool) {
	var zero T
	v, ok := fc.params[key]
	if !ok {
		return zero, false
	}
	t, ok := v.(T)
	return t, ok
}

// SetParam stores a typed value in the per-flow param bag.
func SetParam[T any](fc *FlowCtx, key string, v T) { fc.params[key] = v }

// SharedStore is the cross-host, concurrency-safe param store carried by a
// Runner. Per-host flows publish here for later flows (e.g. exits → relay).
type SharedStore struct {
	mu     sync.RWMutex
	params map[string]any
}

// NewSharedStore returns an empty store.
func NewSharedStore() *SharedStore { return &SharedStore{params: map[string]any{}} }

// Set stores a value.
func (s *SharedStore) Set(key string, v any) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.params[key] = v
}

// Get reads a value.
func (s *SharedStore) Get(key string) (any, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	v, ok := s.params[key]
	return v, ok
}

// SharedParam reads a typed value from a SharedStore.
func SharedParam[T any](s *SharedStore, key string) (T, bool) {
	var zero T
	s.mu.RLock()
	defer s.mu.RUnlock()
	v, ok := s.params[key]
	if !ok {
		return zero, false
	}
	t, ok := v.(T)
	return t, ok
}

// SetSharedParam stores a typed value in a SharedStore.
func SetSharedParam[T any](s *SharedStore, key string, v T) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.params[key] = v
}


1	package legatus
2
3	import (
4	"context"
5	"io"
6	"log/slog"
7	"sync"
8	)
9
10	// FlowCtx threads through the steps of one flow execution on one host. Its param
11	// bag is private to this execution, so steps pass data forward without locking
12	// (IV3). Cross-host data goes through Shared.
13	type FlowCtx struct {
14	ctx context.Context
15	host *Host
16	conn Conn
17	log *slog.Logger
18	dryRun bool
19	prompter Prompter
20	shared *SharedStore
21	params map[string]any
22	changed map[string]bool
23	out io.Writer
24	}
25
26	// FlowCtxOption configures a FlowCtx.
27	type FlowCtxOption func(*FlowCtx)
28
29	// WithFlowLogger sets the per-flow logger.
30	func WithFlowLogger(l *slog.Logger) FlowCtxOption {
31	return func(fc *FlowCtx) {
32	if l != nil {
33	fc.log = l
34	}
35	}
36	}
37
38	// WithFlowDryRun toggles dry-run.
39	func WithFlowDryRun(d bool) FlowCtxOption { return func(fc *FlowCtx) { fc.dryRun = d } }
40
41	// WithFlowPrompter sets the prompter.
42	func WithFlowPrompter(p Prompter) FlowCtxOption {
43	return func(fc *FlowCtx) {
44	if p != nil {
45	fc.prompter = p
46	}
47	}
48	}
49
50	// WithFlowShared sets the cross-host shared store.
51	func WithFlowShared(s *SharedStore) FlowCtxOption {
52	return func(fc *FlowCtx) {
53	if s != nil {
54	fc.shared = s
55	}
56	}
57	}
58
59	// WithFlowOutput sets the human-facing output sink steps stream to (Out()).
60	func WithFlowOutput(w io.Writer) FlowCtxOption {
61	return func(fc *FlowCtx) {
62	if w != nil {
63	fc.out = w
64	}
65	}
66	}
67
68	// NewFlowCtx builds a FlowCtx. The Runner uses it; step tests use it directly.
69	func NewFlowCtx(ctx context.Context, h Host, c Conn, opts ...FlowCtxOption) FlowCtx {
70	fc := &FlowCtx{
71	ctx: ctx,
72	host: h,
73	conn: c,
74	log: slog.Default(),
75	prompter: DenyPrompter{},
76	shared: NewSharedStore(),
77	params: map[string]any{},
78	changed: map[string]bool{},
79	out: io.Discard,
80	}
81	for _, o := range opts {
82	o(fc)
83	}
84	return fc
85	}
86
87	// Ctx returns the execution context.
88	func (fc *FlowCtx) Ctx() context.Context { return fc.ctx }
89
90	// Host returns the target host.
91	func (fc FlowCtx) Host() Host { return fc.host }
92
93	// Conn returns the live connection to the host.
94	func (fc *FlowCtx) Conn() Conn { return fc.conn }
95
96	// Out returns the human-facing output sink. The engine owns it; steps stream
97	// command output here (never os.Stdout) when asked. Never nil — io.Discard by
98	// default, so an unconfigured Runner silently discards.
99	func (fc *FlowCtx) Out() io.Writer { return fc.out }
100
101	// Log returns the per-flow logger, pre-tagged by the Runner with host and step.
102	func (fc FlowCtx) Log() slog.Logger { return fc.log }
103
104	// DryRun reports whether this is a dry run.
105	func (fc *FlowCtx) DryRun() bool { return fc.dryRun }
106
107	// Prompter returns the injected prompter (non-interactive by default).
108	func (fc *FlowCtx) Prompter() Prompter { return fc.prompter }
109
110	// Shared returns the cross-host store.
111	func (fc FlowCtx) Shared() SharedStore { return fc.shared }
112
113	// Set stores a value in the per-flow param bag.
114	func (fc *FlowCtx) Set(key string, v any) { fc.params[key] = v }
115
116	// Get reads a value from the per-flow param bag.
117	func (fc *FlowCtx) Get(key string) (any, bool) {
118	v, ok := fc.params[key]
119	return v, ok
120	}
121
122	// setChanged records, by step name, whether a step changed (or would change in
123	// dry-run). The runner calls it after each step resolves, before the next step's
124	// Check. Duplicate names OR-accumulate, so two same-named steps collapse to "any
125	// of them changed".
126	func (fc *FlowCtx) setChanged(name string, did bool) {
127	fc.changed[name] = fc.changed[name] \|\| did
128	}
129
130	// Changed reports whether the named earlier step changed (or would change in
131	// dry-run) during this flow execution on this host. An unrun or unrecorded step
132	// reports false.
133	func (fc *FlowCtx) Changed(name string) bool { return fc.changed[name] }
134
135	// AnyChanged reports whether any of the named earlier steps changed. With no
136	// names, it reports whether any earlier step in this flow changed. Visibility is
137	// limited to steps already executed on this host (never cross-host).
138	func (fc *FlowCtx) AnyChanged(names ...string) bool {
139	if len(names) == 0 {
140	for _, v := range fc.changed {
141	if v {
142	return true
143	}
144	}
145	return false
146	}
147	for _, n := range names {
148	if fc.changed[n] {
149	return true
150	}
151	}
152	return false
153	}
154
155	func (fc *FlowCtx) snapshot() map[string]any {
156	out := make(map[string]any, len(fc.params))
157	for k, v := range fc.params {
158	out[k] = v
159	}
160	return out
161	}
162
163	// Param reads a typed value from the per-flow param bag.
164	func Param[T any](fc *FlowCtx, key string) (T, bool) {
165	var zero T
166	v, ok := fc.params[key]
167	if !ok {
168	return zero, false
169	}
170	t, ok := v.(T)
171	return t, ok
172	}
173
174	// SetParam stores a typed value in the per-flow param bag.
175	func SetParam[T any](fc *FlowCtx, key string, v T) { fc.params[key] = v }
176
177	// SharedStore is the cross-host, concurrency-safe param store carried by a
178	// Runner. Per-host flows publish here for later flows (e.g. exits → relay).
179	type SharedStore struct {
180	mu sync.RWMutex
181	params map[string]any
182	}
183
184	// NewSharedStore returns an empty store.
185	func NewSharedStore() *SharedStore { return &SharedStore{params: map[string]any{}} }
186
187	// Set stores a value.
188	func (s *SharedStore) Set(key string, v any) {
189	s.mu.Lock()
190	defer s.mu.Unlock()
191	s.params[key] = v
192	}
193
194	// Get reads a value.
195	func (s *SharedStore) Get(key string) (any, bool) {
196	s.mu.RLock()
197	defer s.mu.RUnlock()
198	v, ok := s.params[key]
199	return v, ok
200	}
201
202	// SharedParam reads a typed value from a SharedStore.
203	func SharedParam[T any](s *SharedStore, key string) (T, bool) {
204	var zero T
205	s.mu.RLock()
206	defer s.mu.RUnlock()
207	v, ok := s.params[key]
208	if !ok {
209	return zero, false
210	}
211	t, ok := v.(T)
212	return t, ok
213	}
214
215	// SetSharedParam stores a typed value in a SharedStore.
216	func SetSharedParam[T any](s *SharedStore, key string, v T) {
217	s.mu.Lock()
218	defer s.mu.Unlock()
219	s.params[key] = v
220	}
221

flowctx.go

Source Files