Documentation
¶
Overview ¶
Package flow is tiny's local implementation of the platform FlowService — the gRPC/gRPC-web API the browser editor talks to. It's backed by the cluster's TinyFlow/TinyNode CRDs (via the SDK's resource.Manager) and the shared SDK graph helpers (module/pkg/utils, module/pkg/schema), the same code the hosted platform calls. Only the RPCs the local editor needs are implemented; everything else falls through to UnimplementedFlowServiceServer (platform-only features — LLM prompt, revision history, registry browse).
Index ¶
- func Serve(ctx context.Context, addr string, svc *Service, activeProject string, ...) error
- type ActivityBus
- type Service
- func (s *Service) AcquireFlowLock(ctx context.Context, req *platform.AcquireFlowLockRequest) (*platform.AcquireFlowLockResponse, error)
- func (s *Service) CreateFlow(ctx context.Context, req *platform.CreateFlowRequest) (*platform.Nil, error)
- func (s *Service) GetComponents(ctx context.Context, req *platform.GetComponentsRequest) (*platform.GetComponentsResponse, error)
- func (s *Service) GetFlow(ctx context.Context, req *platform.GetFlowRequest) (*platform.GetFlowResponse, error)
- func (s *Service) GetFlowList(ctx context.Context, req *platform.GetFlowListRequest) (*platform.GetFlowListResponse, error)
- func (s *Service) GetFlowStream(req *platform.GetFlowStreamRequest, ...) error
- func (s *Service) InspectNode(ctx context.Context, req *platform.InspectRequest) (*platform.InspectResponse, error)
- func (s *Service) ListScenarios(ctx context.Context, req *platform.ListScenariosRequest) (*platform.ListScenariosResponse, error)
- func (s *Service) PreviewEdgeMapping(ctx context.Context, req *platform.PreviewEdgeMappingRequest) (*platform.PreviewEdgeMappingResponse, error)
- func (s *Service) ReleaseFlowLock(ctx context.Context, req *platform.ReleaseFlowLockRequest) (*platform.Nil, error)
- func (s *Service) RenameFlow(ctx context.Context, req *platform.RenameFlowRequest) (*platform.Nil, error)
- func (s *Service) RunAction(ctx context.Context, req *platform.RunActionRequest) (*platform.Nil, error)
- func (s *Service) RunExpression(ctx context.Context, req *platform.RunExpressionRequest) (*platform.RunExpressionResponse, error)
- func (s *Service) SaveFlow(ctx context.Context, req *platform.SaveFlowRequest) (*platform.SaveFlowResponse, error)
- func (s *Service) SaveFlowMeta(ctx context.Context, req *platform.SaveFlowMetaRequest) (*platform.Nil, error)
- func (s *Service) SetSignalSender(sig sdktools.SignalSender)
- func (s *Service) SetTraceReader(t traceSource)
- func (s *Service) UndeployFlow(ctx context.Context, req *platform.UndeployFlowRequest) (*platform.Nil, error)
- type Tunnel
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Serve ¶
func Serve(ctx context.Context, addr string, svc *Service, activeProject string, bus *ActivityBus, staticFS http.Handler) error
Serve runs the FlowService as a gRPC-web endpoint on addr (e.g. "127.0.0.1:7775") until ctx is cancelled. The editor's Connect-ES createGrpcWebTransport client talks to it directly — same wire protocol the hosted platform serves — with CORS opened for the localhost browser.
activeProject is the session's fixed project (one per session), surfaced at /api/session so the SPA knows which project to open without a switcher.
staticFS, when non-nil, serves the editor SPA for any non-gRPC request; nil leaves the endpoint gRPC-web-only.
Types ¶
type ActivityBus ¶
type ActivityBus struct {
// contains filtered or unexported fields
}
ActivityBus is a tiny in-process pub/sub for agent activity. tiny's MCP server publishes an event on every tool call; the editor's WorkspaceActivityService streams them to the dashboard's Activity feed. It keeps a small backlog so a feed that connects mid-session sees recent events.
func NewActivityBus ¶
func NewActivityBus() *ActivityBus
func (*ActivityBus) Publish ¶
func (b *ActivityBus) Publish(kind string)
Publish records a bare event by kind, for activity that carries no payload.
func (*ActivityBus) PublishToolEnded ¶ added in v0.4.41
func (b *ActivityBus) PublishToolEnded(tool string, success bool, errMsg string, duration time.Duration)
PublishToolEnded announces a tool call finishing, WITH its outcome.
The payload matters: the feed reads success/tool off it, so publishing a bare kind made every finished call render as an unnamed "tool failed" — the events were fine, the verdict was missing.
func (*ActivityBus) PublishToolStarted ¶ added in v0.4.41
func (b *ActivityBus) PublishToolStarted(tool string)
PublishToolStarted announces a tool call beginning.
type Service ¶
type Service struct {
platform.UnimplementedFlowServiceServer
// contains filtered or unexported fields
}
Service implements platform.FlowServiceServer against a local cluster.
func NewService ¶
NewService binds the service to one cluster + namespace.
func (*Service) AcquireFlowLock ¶
func (s *Service) AcquireFlowLock(ctx context.Context, req *platform.AcquireFlowLockRequest) (*platform.AcquireFlowLockResponse, error)
AcquireFlowLock always grants the lock: a single-user local editor has no contention, and the frontend gates editable rendering on this response.
func (*Service) CreateFlow ¶
func (s *Service) CreateFlow(ctx context.Context, req *platform.CreateFlowRequest) (*platform.Nil, error)
CreateFlow adds a new flow (layer) to the project.
func (*Service) GetComponents ¶
func (s *Service) GetComponents(ctx context.Context, req *platform.GetComponentsRequest) (*platform.GetComponentsResponse, error)
GetComponents lists the installed modules' components for the editor's add-component palette. Each item carries a minimal node graph (module + component); the operator fills in ports on reconcile after the node is created, and the stream updates the editor.
func (*Service) GetFlow ¶
func (s *Service) GetFlow(ctx context.Context, req *platform.GetFlowRequest) (*platform.GetFlowResponse, error)
GetFlow returns flow metadata from the TinyFlow CR. The graph itself is streamed by GetFlowStream — the editor reads only ID/ResourceName/Meta here.
func (*Service) GetFlowList ¶
func (s *Service) GetFlowList(ctx context.Context, req *platform.GetFlowListRequest) (*platform.GetFlowListResponse, error)
GetFlowList lists the project's flows (the layers) for the flow switcher.
func (*Service) GetFlowStream ¶
func (s *Service) GetFlowStream(req *platform.GetFlowStreamRequest, stream grpc.ServerStreamingServer[platform.GetFlowStreamResponse]) error
GetFlowStream is the render path: it streams the flow's nodes and edges to the canvas, then keeps the graph live as the cluster changes.
This is a stripped buildGraphEvents — the hosted platform overlays otel stats, redis logs, revision notices, and lock state onto the same stream; all of those degrade gracefully, so locally we ship just the graph: WatchNodes → SDK graph maps → node/edge upserts (ADDED/MODIFIED), delete events for elements that disappear, and a 2s heartbeat. The heavy work — schema overlay and edge validation — is the SDK's, identical to the platform.
func (*Service) InspectNode ¶
func (s *Service) InspectNode(ctx context.Context, req *platform.InspectRequest) (*platform.InspectResponse, error)
InspectNode returns a port's data shape — the example/simulated data the editor shows in a node's Debug/Config tab and uses to preview edge mappings. It's backed by the same PortInspector the MCP tools use: it reads the node's reconciled port schema + configuration and returns the example data as JSON, which the editor reads from response.Data.
This is the narrower local inspector (the node's own ports enriched with its own _settings overlay), not the platform's whole-graph simulation — enough for the editor to populate the inspector and edge preview.
func (*Service) ListScenarios ¶
func (s *Service) ListScenarios(ctx context.Context, req *platform.ListScenariosRequest) (*platform.ListScenariosResponse, error)
ListScenarios returns the flow's scenarios for the editor's scenario switcher. Local flows have none beyond the implicit Default, so this returns empty — enough to stop the switcher's on-mount call from erroring.
func (*Service) PreviewEdgeMapping ¶
func (s *Service) PreviewEdgeMapping(ctx context.Context, req *platform.PreviewEdgeMappingRequest) (*platform.PreviewEdgeMappingResponse, error)
PreviewEdgeMapping applies an edge's configuration mapping to sample source data and returns the mapped result — the live preview in the edge-config panel as you type a mapping. Also pure SDK evaluation.
func (*Service) ReleaseFlowLock ¶
func (s *Service) ReleaseFlowLock(ctx context.Context, req *platform.ReleaseFlowLockRequest) (*platform.Nil, error)
ReleaseFlowLock is a no-op locally.
func (*Service) RenameFlow ¶ added in v0.4.41
func (s *Service) RenameFlow(ctx context.Context, req *platform.RenameFlowRequest) (*platform.Nil, error)
RenameFlow updates a flow's human-readable name. That name lives in the TinyFlow's description annotation (createFlow writes it there and the editor reads it), so a rename is a one-field annotation update — no resources move, the flow's resource name and node bindings are untouched.
The editor sends the new name as JSON in RenameForm.Data under "newName", matching the platform's form, so the same dialog drives both. Without this handler the menu's Rename action hit an unregistered method and failed silently.
func (*Service) RunAction ¶
func (s *Service) RunAction(ctx context.Context, req *platform.RunActionRequest) (*platform.Nil, error)
RunAction fires data into a node's port — the editor's "run"/test action on a control widget. Delegates to the injected NATS-backed signal sender.
func (*Service) RunExpression ¶
func (s *Service) RunExpression(ctx context.Context, req *platform.RunExpressionRequest) (*platform.RunExpressionResponse, error)
RunExpression evaluates an ajson expression against sample data and validates the result against a schema — the expression testing + edge-mapping checks in the editor's config panel. Pure SDK evaluation, no cluster access, so it's a direct passthrough to the shared evaluator the platform uses.
func (*Service) SaveFlow ¶
func (s *Service) SaveFlow(ctx context.Context, req *platform.SaveFlowRequest) (*platform.SaveFlowResponse, error)
SaveFlow reconciles the editor's whole-graph save into cluster CRs. The editor sends every element of the flow; SaveFlow diffs that against the TinyNodes this flow owns and creates / updates / deletes to match.
It is deliberately conservative: it only ever touches nodes labelled for THIS flow. Nodes owned by other flows (shared into this one) are left untouched even if they appear in the graph, and are never deleted.
func (*Service) SaveFlowMeta ¶
func (s *Service) SaveFlowMeta(ctx context.Context, req *platform.SaveFlowMetaRequest) (*platform.Nil, error)
SaveFlowMeta persists the editor viewport (x/y/zoom) as an annotation on the TinyFlow so it restores on reload.
func (*Service) SetSignalSender ¶
func (s *Service) SetSignalSender(sig sdktools.SignalSender)
SetSignalSender wires the node-fire capability (RunAction). The host passes the same NATS-backed sender the MCP tools use.
func (*Service) SetTraceReader ¶
func (s *Service) SetTraceReader(t traceSource)
SetTraceReader wires the Statistics (traces) capability. The host passes the same otel-collector trace reader the MCP get_traces tools use.
func (*Service) UndeployFlow ¶
func (s *Service) UndeployFlow(ctx context.Context, req *platform.UndeployFlowRequest) (*platform.Nil, error)
UndeployFlow deletes a flow (layer) and all the TinyNodes that belong to it — the editor's flow delete/undeploy action. Locally undeploy IS delete: there's no separate deployed/undeployed state, so it removes the flow outright.
type Tunnel ¶
type Tunnel struct {
// contains filtered or unexported fields
}
Tunnel keeps the user's localhost in sync with the servers running inside the cluster. A server component (e.g. http_server) binds a random port INSIDE its module's controller-manager pod and reports it as a `listenAddr` of http://localhost:<port> on the node's _control port. That address is the pod's loopback — nothing on the user's machine can reach it. Tunnel watches for those listen addresses and port-forwards each pod:<port> to 127.0.0.1:<port> (the same port), so the URL the editor already shows just works in the browser. Forwards start when a server comes up and stop when it goes away or its pod is replaced.
func NewTunnel ¶
NewTunnel builds a Tunnel over the same cluster/namespace the editor serves. It creates its own kube client (scheme-aware, lists both TinyNodes and Pods).
Source Files