protocol

type Catalog ¶

type Catalog interface {
	// ListFlows returns every registered graph-family flow with its
	// aggregate run metrics over the trailing 24h window, scoped to the
	// caller's identity (admin fans across tenants).
	ListFlows(ctx context.Context, id identity.Identity, adminScoped bool) ([]prototypes.Flow, error)
	// DescribeFlow returns a single flow's full engine-graph
	// description. An unknown id returns ErrNotFound.
	DescribeFlow(ctx context.Context, id identity.Identity, adminScoped bool, flowID string) (prototypes.FlowDescription, error)
	// ListRuns returns a flow's run history, newest first, scoped to the
	// caller's identity (admin fans across tenants when crossTenant is
	// requested). An unknown flow id returns ErrNotFound.
	ListRuns(ctx context.Context, id identity.Identity, adminScoped bool, flowID string, tenants []string) ([]prototypes.FlowRun, error)
	// DescribeRun returns a single run's per-node timeline + output
	// reference. An unknown run id returns ErrNotFound. The
	// implementation routes heavy outputs by-reference.
	DescribeRun(ctx context.Context, id identity.Identity, adminScoped bool, runID string) (prototypes.FlowRunDescription, error)
	// FlowMetrics returns a flow's time-bucketed sparkline aggregates
	// over the requested window. An unknown flow id returns ErrNotFound.
	FlowMetrics(ctx context.Context, id identity.Identity, adminScoped bool, flowID string, window, bucket time.Duration) (prototypes.FlowMetrics, error)
}

Catalog is the read-only source-of-truth seam the Flows-page Surface depends on. The concrete implementation is the runtime's flow registry + run-history store; tests inject a deterministic fixture.

Every method takes the caller identity + an adminScoped flag. The implementation MUST scope run history to the caller's tenant unless adminScoped is true. Flow definitions themselves are tenant-agnostic (they are descriptors registered at agent-definition time), so the catalog listing is not tenant-filtered — but the per-flow run aggregates ARE (a non-admin caller sees only their own tenant's run counts).

type CatalogOption ¶

type CatalogOption func(*RegistryCatalog)

CatalogOption configures a RegistryCatalog at construction.

func WithCatalogClock(now func() time.Time) CatalogOption

type FuncInvoker ¶

type FuncInvoker struct {
	// contains filtered or unexported fields
}

FuncInvoker is the production Invoker implementation. It adapts a runtime-supplied LaunchFunc onto the Invoker interface. It is NOT a test stub (CLAUDE.md §13): the binary binds the LaunchFunc to the real `start` path; the FuncInvoker is the thin adapter that keeps the Flows-page surface decoupled from the task subsystem.

Concurrent reuse: the FuncInvoker is a compiled artifact — the LaunchFunc is set once at construction and never mutated.

func NewFuncInvoker(launch LaunchFunc, registry *flow.Registry) (*FuncInvoker, error)

func (i *FuncInvoker) Invoke(ctx context.Context, id identity.Identity, flowID string, inputs map[string]any) (prototypes.FlowRunResponse, error)

type Invoker ¶

type Invoker interface {
	// Invoke launches a one-shot run of flowID with the supplied inputs
	// under the caller's identity. It returns the accepted run's
	// identifier + start time. An unknown flow id returns ErrNotFound; a
	// malformed input form returns ErrInvalidRequest.
	Invoke(ctx context.Context, id identity.Identity, flowID string, inputs map[string]any) (prototypes.FlowRunResponse, error)
}

Invoker launches a one-shot run of a registered flow. The concrete implementation wraps the task registry's `start` path; tests inject a deterministic fixture. Invoke runs under the caller's identity — the launched run inherits the (tenant, user, session) triple.

type LaunchFunc ¶

type LaunchFunc func(ctx context.Context, id identity.Identity, flowID string, inputs map[string]any) (runID string, startedAt time.Time, err error)

LaunchFunc is the runtime-supplied launcher the FuncInvoker wraps. It launches a one-shot run of the named flow under the caller's identity and returns the accepted run's identifier + start time. The runtime binds this to its real task-registry `start` path — the FuncInvoker never reaches the task registry directly, so the Flows- page surface stays decoupled from the task subsystem's concrete type.

A LaunchFunc that targets an unknown flow returns an error wrapping ErrNotFound; a malformed input form returns an error wrapping ErrInvalidRequest. Any other failure is wrapped as ErrRuntime by the Surface's classifyCatalogErr.

type RegistryCatalog ¶

type RegistryCatalog struct {
	// contains filtered or unexported fields
}

RegistryCatalog is the production Catalog implementation. It projects the Console Flows-page wire shapes from a *flow.Registry — the runtime's source-of-truth for registered flows + run history. It is NOT a test stub (CLAUDE.md §13): it is the real catalog backed by the real registry; the binary wires it at boot.

Heavy run outputs are routed by-reference through an ArtifactStore: a run whose final output meets or exceeds the configured heavy-content threshold ships a FlowArtifactRef, never inline bytes.

Concurrent reuse: the RegistryCatalog is a compiled artifact — registry / store / threshold are set once at construction. Every method reads through the registry's own RWMutex; the catalog holds no per-call state.

func NewRegistryCatalog(registry *flow.Registry, store artifacts.ArtifactStore, threshold int, opts ...CatalogOption) (*RegistryCatalog, error)

func (c *RegistryCatalog) DescribeFlow(ctx context.Context, id identity.Identity, adminScoped bool, flowID string) (prototypes.FlowDescription, error)

func (c *RegistryCatalog) DescribeRun(ctx context.Context, id identity.Identity, adminScoped bool, runID string) (prototypes.FlowRunDescription, error)

func (c *RegistryCatalog) FlowMetrics(ctx context.Context, id identity.Identity, adminScoped bool, flowID string, window, bucket time.Duration) (prototypes.FlowMetrics, error)

func (c *RegistryCatalog) ListFlows(ctx context.Context, id identity.Identity, adminScoped bool) ([]prototypes.Flow, error)

func (c *RegistryCatalog) ListRuns(ctx context.Context, id identity.Identity, adminScoped bool, flowID string, tenants []string) ([]prototypes.FlowRun, error)

type Surface ¶

type Surface struct {
	// contains filtered or unexported fields
}

Surface is the transport-agnostic Flows-page Protocol surface. It is a compiled artifact: catalog + invoker are set once at construction and never mutated; ServeMethod holds no per-call state.

func NewSurface(catalog Catalog, invoker Invoker) (*Surface, error)

func (s *Surface) Describe(ctx context.Context, req prototypes.FlowDescribeRequest, adminScoped bool) (prototypes.FlowDescription, error)

func (s *Surface) List(ctx context.Context, req prototypes.FlowListRequest, adminScoped bool) (prototypes.FlowListResponse, error)

func (s *Surface) Metrics(ctx context.Context, req prototypes.FlowMetricsRequest, adminScoped bool) (prototypes.FlowMetrics, error)

func (s *Surface) Run(ctx context.Context, req prototypes.FlowRunRequest, adminScoped bool) (prototypes.FlowRunResponse, error)

func (s *Surface) RunsDescribe(ctx context.Context, req prototypes.FlowRunDescribeRequest, adminScoped bool) (prototypes.FlowRunDescription, error)

func (s *Surface) RunsList(ctx context.Context, req prototypes.FlowRunsListRequest, adminScoped bool) (prototypes.FlowRunsListResponse, error)