service

type BaseOperator ¶

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

BaseOperator defines a template Notify flow: 1) update close-state on closed signal 2) delegate data handling to onData hook 3) attempt finish sequence if all inputs are closed

func (o *BaseOperator) IsFinished() bool

func (o *BaseOperator) Notify(ctx context.Context, signal DataflowSignal) error

type DataflowAnalysisNodePersistParams ¶

type DataflowAnalysisNodePersistParams struct {
	AnalysisID      int64
	NodeID          string
	InputHash       string
	NodeName        string
	SampleID        string
	ScriptID        string
	InputsPatterns  map[string]any
	OutputPatterns  map[string]any
	Params          map[string]any
	ResolvedInputs  map[string]any
	ResolvedOutputs map[string]any
	UpstreamIDs     []string
	DownstreamIDs   []string
	Executor        string
	Retry           int
	MaxRetry        int
	RerunReason     string
	Status          string
	SubmitReason    string
	WorkspaceDir    string
	OutputDir       string
	CommandPath     string
	ParamsPath      string
	LogPath         string
}

DataflowAnalysisNodePersistParams is a normalized payload that contains the fields needed to create an analysis_node record.

It is assembled at the runtime submit boundary and will be consumed by a persistent runtime implementation in the next step.

type DataflowChannel ¶

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

DataflowChannel models a Nextflow-like channel lifecycle: open -> emit* -> close.

func (ch *DataflowChannel) Close(ctx context.Context) error

func (ch *DataflowChannel) Emit(ctx context.Context, value any) error

func (ch *DataflowChannel) ID() string

func (ch *DataflowChannel) IsClosed() bool

func (ch *DataflowChannel) Subscribe(op DataflowOperator)

type DataflowChannelSpec ¶

type DataflowChannelSpec struct {
	ChannelID  string
	FromNodeID string
	ToNodeID   string
	FromPort   string
	ToPort     string
}

DataflowChannelSpec represents a logical edge/channel in the dataflow graph.

type DataflowGraphSpec ¶

type DataflowGraphSpec struct {
	AnalysisID int64
	Processes  []DataflowProcessSpec
	Channels   []DataflowChannelSpec
}

DataflowGraphSpec is a V3 planning view converted from dag_definition. It is intentionally lightweight for the first framework milestone.

type DataflowOperator ¶

type DataflowOperator interface {
	Notify(ctx context.Context, signal DataflowSignal) error
}

DataflowOperator receives channel signals and decides whether a process instance should be materialized.

type DataflowProcessRunRequest ¶

type DataflowProcessRunRequest struct {
	AnalysisID int64
	NodeID     string
	Inputs     map[string]any
	Reason     string
}

type DataflowProcessRuntime ¶

type DataflowProcessRuntime interface {
	SubmitProcessInstance(ctx context.Context, req DataflowProcessRunRequest) error
}

DataflowProcessRuntime is the bridge from operator decisions to concrete process launch.

type DataflowProcessSpec ¶

type DataflowProcessSpec struct {
	NodeID       string
	NodeName     string
	SampleID     string
	ScriptID     string
	InputKeys    []string
	UpstreamIDs  []string
	Downstream   []string
	Inputs       map[string]any
	Outputs      map[string]any
	Params       map[string]any
	ResolvedIn   map[string]any
	ResolvedOut  map[string]any
	Executor     string
	Retry        int
	MaxRetry     int
	RerunReason  string
	OperatorType string
	ScatterField string
	ScatterMode  string
	GatherField  string
	GatherMode   string
}

DataflowProcessSpec represents a process template (not a persisted node instance).

type DataflowSignal ¶

type DataflowSignal struct {
	ChannelID string
	Value     any
	Closed    bool
}

DataflowSignal is the message exchanged through channels/operators.

type GatherOperator ¶

type GatherOperator struct {
	*BaseOperator
	// contains filtered or unexported fields
}

GatherOperator aggregates one field according to gather mode and emits once when all upstream channels are closed.

type InputOperator ¶

type InputOperator struct {
	*BaseOperator
	// contains filtered or unexported fields
}

InputOperator is the default operator for regular inputs. It collects values from one or more upstream channels and materializes a process instance when all required inputs have at least one value.

This is the V3 equivalent of: - cache.Add(v) - if Ready() { runtime.SubmitTask(...) }

type ScatterOperator ¶

type ScatterOperator struct {
	*BaseOperator
	// contains filtered or unexported fields
}

ScatterOperator expands one input field according to scatter mode. Supported modes: each, list.