Domain Logic¶
Strategy plan AST, operations, and validation. Pure domain logic with no I/O or framework dependencies. Used by the agent, serialization, and tools.
Overview¶
Strategy AST — Types for plans:
PlanStepNode,StrategyAST, combine operators. Recursive structure for search/transform/combine steps.Strategy Operations — Combine operators, colocation params, WDK operator mapping.
Strategy Validation — Validate against WDK constraints; emit field paths.
Parameters — Specs, normalization, canonicalization, vocabulary handling.
Strategy AST¶
Purpose: AST types for strategy plans. PlanStepNode is recursive (search,
transform, combine). StrategyAST wraps root + metadata. Used by the agent
and serialization layer.
Key types: PlanStepNode, StrategyAST
AST node types for strategy representation (WDK-aligned, untyped tree).
- class veupath_chatbot.domain.strategy.ast.StepTreeNode(step_id, primary_input=None, secondary_input=None)[source]¶
Bases:
objectNode in a WDK step tree.
Represents a single step with optional primary (and for combines, secondary) input references. Used to build the
stepTreepayload for WDK strategy creation. Pure data structure with no I/O.
- veupath_chatbot.domain.strategy.ast.generate_step_id()[source]¶
Generate a unique step ID.
- Return type:
- veupath_chatbot.domain.strategy.ast.parse_filters(raw)[source]¶
Parse a list of step filters from raw JSON data.
- Return type:
- veupath_chatbot.domain.strategy.ast.parse_analyses(raw)[source]¶
Parse a list of step analyses from raw JSON data.
- Return type:
- veupath_chatbot.domain.strategy.ast.parse_reports(raw)[source]¶
Parse a list of step reports from raw JSON data.
- Return type:
- veupath_chatbot.domain.strategy.ast.parse_colocation_params(raw)[source]¶
Parse colocation parameters from raw JSON data.
- Return type:
ColocationParams | None
- class veupath_chatbot.domain.strategy.ast.StepFilter(name, value, disabled=False)[source]¶
Bases:
objectFilter applied to a step’s result.
- __init__(name, value, disabled=False)¶
- class veupath_chatbot.domain.strategy.ast.StepAnalysis(analysis_type, parameters=<factory>, custom_name=None)[source]¶
Bases:
objectAnalysis configuration for a step.
- parameters: JSONObject¶
- __init__(analysis_type, parameters=<factory>, custom_name=None)¶
- class veupath_chatbot.domain.strategy.ast.StepReport(report_name='standard', config=<factory>)[source]¶
Bases:
objectReport request attached to a step.
- config: JSONObject¶
- __init__(report_name='standard', config=<factory>)¶
- class veupath_chatbot.domain.strategy.ast.PlanStepNode(search_name, parameters=<factory>, primary_input=None, secondary_input=None, operator=None, colocation_params=None, display_name=None, filters=<factory>, analyses=<factory>, reports=<factory>, wdk_weight=None, id=<factory>)[source]¶
Bases:
objectUntyped recursive strategy node.
Kind is inferred from structure: - combine: primary_input and secondary_input - transform: primary_input only - search: no inputs
- parameters: JSONObject¶
- primary_input: PlanStepNode | None = None¶
- secondary_input: PlanStepNode | None = None¶
- colocation_params: ColocationParams | None = None¶
- filters: list[StepFilter]¶
- analyses: list[StepAnalysis]¶
- reports: list[StepReport]¶
- __init__(search_name, parameters=<factory>, primary_input=None, secondary_input=None, operator=None, colocation_params=None, display_name=None, filters=<factory>, analyses=<factory>, reports=<factory>, wdk_weight=None, id=<factory>)¶
- class veupath_chatbot.domain.strategy.ast.StrategyAST(record_type, root, name=None, description=None, metadata=None)[source]¶
Bases:
objectComplete strategy represented as an AST.
- root: PlanStepNode¶
- metadata: JSONObject | None = None¶
- get_step_by_id(step_id)[source]¶
Find a step by its ID.
- Parameters:
step_id (str) – Step identifier.
- Return type:
PlanStepNode | None
- __init__(record_type, root, name=None, description=None, metadata=None)¶
- veupath_chatbot.domain.strategy.ast.from_dict(data)[source]¶
Parse strategy from dictionary representation.
- Parameters:
data (JSONObject) – Data dict.
- Return type:
Strategy Operations¶
Purpose: Operations on the strategy AST: combine operators, colocation parameters, WDK operator mapping. Used when building and manipulating plans.
Key classes: CombineOp, ColocationParams
Key functions: get_wdk_operator(), parse_op()
Combine operations for strategy building.
Canonical set (matches WDK BooleanOperator): INTERSECT, MINUS, RMINUS, LONLY, RONLY, COLOCATE, UNION. LONLY = left only (same as MINUS), RONLY = right only (same as RMINUS); we keep both for round-trip fidelity with WDK.
- class veupath_chatbot.domain.strategy.ops.CombineOp(*values)[source]¶
Bases:
StrEnumSet operations for combining two step results.
- INTERSECT = 'INTERSECT'¶
- MINUS = 'MINUS'¶
- RMINUS = 'RMINUS'¶
- LONLY = 'LONLY'¶
- RONLY = 'RONLY'¶
- COLOCATE = 'COLOCATE'¶
- UNION = 'UNION'¶
- class veupath_chatbot.domain.strategy.ops.ColocationParams(upstream=0, downstream=0, strand='both')[source]¶
Bases:
objectParameters for colocation operator.
- __init__(upstream=0, downstream=0, strand='both')¶
- veupath_chatbot.domain.strategy.ops.get_wdk_operator(op)[source]¶
Get WDK boolean operator name.
Since enum values now match WDK values directly, this simply returns
op.value(with a guard for COLOCATE which is not a boolean operator).- Parameters:
op (CombineOp) – Combine operator.
- Returns:
WDK boolean operator name.
- Raises:
ValueError – If op is COLOCATE.
- Return type:
- veupath_chatbot.domain.strategy.ops.parse_op(value)[source]¶
Parse operator from string value.
- Parameters:
value (str) – String value to parse.
- Returns:
Parsed combine operator.
- Raises:
ValueError – If value is empty or unknown.
- Return type:
Strategy Validation¶
Purpose: Validate plans against WDK constraints: required parameters, valid
search names, step structure. Emits ValidationError with field paths for
UI display.
Key function: validate_strategy()
Validation for strategy DSL.
- class veupath_chatbot.domain.strategy.validate.StepValidationIssue(path, message, code)[source]¶
Bases:
objectA single validation issue found during step/strategy validation.
- __init__(path, message, code)¶
- class veupath_chatbot.domain.strategy.validate.ValidationResult(valid, errors)[source]¶
Bases:
objectResult of validation.
- errors: list[StepValidationIssue]¶
- classmethod failure(errors)[source]¶
Create a failed result.
- Parameters:
errors (list[StepValidationIssue]) – Validation errors list.
- Return type:
- __init__(valid, errors)¶
- class veupath_chatbot.domain.strategy.validate.StrategyValidator(available_searches=None, available_transforms=None)[source]¶
Bases:
objectValidates strategy AST.
- validate(strategy)[source]¶
Validate a strategy AST.
- Parameters:
strategy (StrategyAST) – Strategy AST.
- Return type:
- veupath_chatbot.domain.strategy.validate.validate_strategy(strategy)[source]¶
Validate a strategy AST with default validator.
- Parameters:
strategy (StrategyAST) – Strategy AST.
- Return type:
Parameters (Domain)¶
Purpose: Parameter specs, normalization, and validation. Vocabulary flattening, canonicalization, and value decoding. Used by catalog and tools.
Key functions (specs): extract_param_specs(), adapt_param_specs()
Key functions (normalize): ParameterNormalizer
Key functions (canonicalize): Value coercion for WDK wire format
Adapters for WDK parameter specifications.
- class veupath_chatbot.domain.parameters.specs.ParamSpecNormalized(name, param_type, allow_empty_value=False, min_selected_count=None, max_selected_count=None, vocabulary=None, count_only_leaves=False, is_number=False, min_value=None, max_value=None, increment=None, max_length=None, display_type='', is_visible=True, group='', dependent_params=(), help=None)[source]¶
Bases:
objectCanonical representation of a WDK parameter spec.
- vocabulary: JSONObject | JSONArray | None = None¶
- __init__(name, param_type, allow_empty_value=False, min_selected_count=None, max_selected_count=None, vocabulary=None, count_only_leaves=False, is_number=False, min_value=None, max_value=None, increment=None, max_length=None, display_type='', is_visible=True, group='', dependent_params=(), help=None)¶
- veupath_chatbot.domain.parameters.specs.unwrap_search_data(details)[source]¶
Normalize WDK/discovery payload shape to the dict that contains parameters.
- Parameters:
details (JSONObject | None) – Search details from WDK/discovery.
- Returns:
Search data dict or None.
- Return type:
JSONObject | None
- veupath_chatbot.domain.parameters.specs.extract_param_specs(payload)[source]¶
Extract parameter spec dicts from a WDK payload.
- Supports the canonical WDK paths (in priority order):
payload["parameters"]– list or dictpayload["paramMap"]– dict (name -> spec)payload["searchConfig"]["parameters"]– list or dictpayload["searchConfig"]["paramMap"]– dict
- Return type:
- veupath_chatbot.domain.parameters.specs.find_input_step_param(specs)[source]¶
- Return type:
str | None
- veupath_chatbot.domain.parameters.specs.find_missing_required_params(param_specs, parameters)[source]¶
Find required parameters that are missing or empty in the given values.
Shared by
validation.pyandparam_validation.pyto keep the required-check logic in a single place.- Parameters:
param_specs (JSONArray) – Raw WDK parameter spec dicts.
parameters (JSONObject) – Parameter values to check.
- Returns:
List of missing required parameter names.
- Return type:
Normalize parameter values into WDK wire format.
Delegates validation and decoding to the shared dispatch chain in
_value_helpers._process_value(), then applies WDK wire formatting:
compound types (multi-pick lists, ranges, filter dicts/lists) are
serialized as JSON strings.
- class veupath_chatbot.domain.parameters.normalize.ParameterNormalizer(specs)[source]¶
Bases:
ParameterValueMixinNormalize parameter values using canonical parameter specs.
Produces WDK wire-safe values: multi-pick lists become JSON strings, ranges become JSON strings, filters become JSON strings. Does NOT expand selections to leaf terms – even when WDK marks a vocabulary as
countOnlyLeaves, WDK handles that expansion itself.- specs: dict[str, ParamSpecNormalized]¶
- __init__(specs)¶
Canonicalize parameter values (API-friendly) using WDK parameter specs.
This module is the counterpart to domain/parameters/normalize:
normalizeproduces WDK wire-safe values (often strings/JSON strings).canonicalizeproduces API-friendly canonical JSON shapes: multi-pick values becomelist[str], scalars become strings, range values become{min, max}, filter values become dict/list.
Delegates validation and decoding to the shared dispatch chain in
_value_helpers._process_value(), then applies canonicalizer-specific
post-processing (FAKE_ALL_SENTINEL rejection, leaf enforcement).
Used at API boundaries (plan normalization, validation) so the frontend can consume stable shapes without re-implementing coercion.
- class veupath_chatbot.domain.parameters.canonicalize.ParameterCanonicalizer(specs)[source]¶
Bases:
ParameterValueMixinCanonicalize parameter values using canonical parameter specs.
- specs: dict[str, ParamSpecNormalized]¶
- __init__(specs)¶
Shared helpers for decoding parameter values.
- veupath_chatbot.domain.parameters._decode_values.parse_json5_value(raw)[source]¶
- Return type:
JSONValue | None
Shared validation and coercion helpers for parameter processing.
Used by both ParameterNormalizer (wire-safe WDK values) and
ParameterCanonicalizer (API-friendly canonical shapes).
The shared dispatch chain lives in ParameterValueMixin._process_value.
Each consumer (normalizer / canonicalizer) calls it and then applies its
own output formatting to the returned ProcessedParam.
- class veupath_chatbot.domain.parameters._value_helpers.ParamKind(*values)[source]¶
Bases:
EnumDiscriminator for the
ProcessedParamtagged union.- MULTI_PICK = 1¶
- SINGLE_PICK = 2¶
- SCALAR = 3¶
- RANGE = 4¶
- FILTER = 5¶
- INPUT_DATASET = 6¶
- UNKNOWN = 7¶
- EMPTY = 8¶
- class veupath_chatbot.domain.parameters._value_helpers.ProcessedParam(kind, value)[source]¶
Bases:
objectIntermediate result from the shared dispatch chain.
kindtells callers what was produced so they can apply output formatting (e.g.json.dumpsfor the wire normalizer, identity for the canonical API formatter).valueholds the decoded, validated, native-Python value:MULTI_PICK ->
list[str]SINGLE_PICK ->
strSCALAR ->
strRANGE ->
dictwithmin/maxkeysFILTER ->
dict | list | strINPUT_DATASET ->
strUNKNOWN -> original
JSONValue(pass-through)EMPTY ->
""
- __init__(kind, value)¶
- class veupath_chatbot.domain.parameters._value_helpers.ParameterValueMixin[source]¶
Bases:
objectShared helpers for
ParameterNormalizerandParameterCanonicalizer.
Vocabulary utilities.
- veupath_chatbot.domain.parameters.vocab_utils.numeric_equivalent(a, b)[source]¶
Check if two string values represent the same number.
Handles precision differences between WDK vocab values (often floats) and imported strategy parameters (often full-precision decimals).
- Return type:
- veupath_chatbot.domain.parameters.vocab_utils.match_vocab_value(*, vocab, param_name, value)[source]¶
Match a user-supplied value against a vocabulary, returning the canonical form.
Tries exact display match, exact value match, then numeric equivalence. Raises
ValidationErrorif no match is found.- Return type:
- veupath_chatbot.domain.parameters.vocab_utils.get_node_term(node)[source]¶
Return the
termstring from a vocab tree node, or None.- Return type:
str | None
- veupath_chatbot.domain.parameters.vocab_utils.get_vocab_children(node)[source]¶
Return typed child nodes from a vocab tree node.
- Return type:
- veupath_chatbot.domain.parameters.vocab_utils.find_vocab_node(root, match)[source]¶
Find a node whose
termordisplayequals match (DFS).- Return type:
JSONObject | None
Strategy Tree Walkers¶
Purpose: Shared tree traversal utilities for strategy trees. Supports both dict-based (raw JSON) and AST-based (typed Pydantic) tree structures. Used by validation, compilation, and analysis.
Shared tree walkers for strategy trees.
Two families of trees appear throughout the codebase:
Dict-based trees – raw WDK
stepTreepayloads orPlanStepNode.to_dict()output. Children live under"primaryInput"and"secondaryInput"keys.AST trees –
PlanStepNodeobjects with.primary_input/.secondary_inputattributes.
This module provides generic, reusable walkers for both so that every call site does not need to re-implement the recursive descent.
- veupath_chatbot.domain.strategy.tree.walk_dict_tree(root, visitor)[source]¶
Pre-order walk of a dict-based step tree.
Calls visitor on every node (the node dict itself), then recurses into
primaryInputandsecondaryInputwhen they are dicts.No-ops silently if root is not a dict.
- veupath_chatbot.domain.strategy.tree.collect_dict_nodes(root)[source]¶
Collect all nodes in a dict-based tree (pre-order).
- Return type:
- veupath_chatbot.domain.strategy.tree.collect_dict_leaves(root)[source]¶
Collect leaf nodes (no primaryInput or secondaryInput) from a dict tree.
- Return type:
- veupath_chatbot.domain.strategy.tree.collect_dict_combine_nodes(root)[source]¶
Collect combine (binary) nodes from a dict tree.
A combine node has both
primaryInputandsecondaryInputas dicts.- Return type:
- veupath_chatbot.domain.strategy.tree.count_dict_nodes(root)[source]¶
Count all nodes in a dict-based tree.
- Return type:
- veupath_chatbot.domain.strategy.tree.map_dict_tree(root, transform)[source]¶
Bottom-up map: apply transform to every node, children first.
Children are replaced with their transformed versions before the parent is passed to transform. The original tree is not mutated; each node dict is shallow-copied before transformation.
- Return type:
- veupath_chatbot.domain.strategy.tree.walk_plan_tree(root, visitor)[source]¶
Pre-order walk of a PlanStepNode AST.
Calls visitor on every node, then recurses into
.primary_inputand.secondary_input.
- veupath_chatbot.domain.strategy.tree.collect_plan_nodes(root)[source]¶
Collect all AST nodes (pre-order).
- Return type:
Strategy — Additional Modules¶
Compile strategy AST to WDK API calls.
- class veupath_chatbot.domain.strategy.compile.StrategyCompilerAPI(*args, **kwargs)[source]¶
Bases:
ProtocolI/O boundary for strategy compilation.
The compiler calls these methods to create WDK steps and datasets. Any object that satisfies this protocol can be injected – the real
StrategyAPIfrom the integrations layer is one such object.- property client: _CompilerClient¶
- async create_step(record_type, search_name, parameters, custom_name=None, wdk_weight=None)[source]¶
- Return type:
- async create_combined_step(primary_step_id, secondary_step_id, boolean_operator, record_type, custom_name=None, wdk_weight=None)[source]¶
- Return type:
- async create_transform_step(input_step_id, transform_name, parameters, record_type='transcript', custom_name=None, wdk_weight=None)[source]¶
- Return type:
- __init__(*args, **kwargs)¶
- class veupath_chatbot.domain.strategy.compile.StepDecoratorAPI(*args, **kwargs)[source]¶
Bases:
ProtocolI/O boundary for post-compilation step decorations (filters, analyses, reports).
- async run_step_analysis(step_id, analysis_type, parameters=None, custom_name=None)[source]¶
- Return type:
- __init__(*args, **kwargs)¶
- class veupath_chatbot.domain.strategy.compile.CompiledStep(local_id, wdk_step_id, step_type, display_name)[source]¶
Bases:
objectA compiled step with WDK step ID.
- __init__(local_id, wdk_step_id, step_type, display_name)¶
- class veupath_chatbot.domain.strategy.compile.CompilationResult(steps, step_tree, root_step_id)[source]¶
Bases:
objectResult of compiling a strategy to WDK.
- steps: list[CompiledStep]¶
- step_tree: StepTreeNode¶
- __init__(steps, step_tree, root_step_id)¶
- class veupath_chatbot.domain.strategy.compile.StrategyCompiler(api, site_id=None, resolve_record_type=True, resolve_search_record_type=None)[source]¶
Bases:
objectCompiles strategy AST to WDK API calls.
- async veupath_chatbot.domain.strategy.compile.apply_step_decorations(strategy, compiled_map, api)[source]¶
Apply filters, analyses, and reports to compiled WDK steps.
Post-compilation step: walks the strategy AST and applies any declared decorations (filters, analyses, reports) to each step’s WDK counterpart.
- async veupath_chatbot.domain.strategy.compile.compile_strategy(strategy, api, site_id=None, resolve_record_type=True, resolve_search_record_type=None)[source]¶
Compile a strategy AST to WDK.
- Return type:
Generate human-readable explanations of strategies.
- veupath_chatbot.domain.strategy.explain.explain_operation(op)[source]¶
Explain what a combine operation does.
Small helpers related to strategy graph metadata.
Stateful strategy session types (in-memory).
These types model the working state while a user (or an AI agent) is building a VEuPathDB strategy during a chat session.
- class veupath_chatbot.domain.strategy.session.StrategyGraph(graph_id, name, site_id)[source]¶
Bases:
objectState for a single strategy graph.
- invalidate_build()[source]¶
Clear WDK build state so stale counts are not shown.
Call after any mutation that changes step semantics (parameters, search_name, operator, delete). The next
build_strategycall will re-populatestep_countsandwdk_step_ids.
- add_step(step)[source]¶
Add a step and maintain the subtree-root set.
The new step becomes a root. If it consumes existing roots as
primary_inputorsecondary_input, those are removed from the root set (they are now internal nodes of the new step’s subtree).- Parameters:
step (PlanStepNode) – Step to add.
- Returns:
Step ID.
- Return type:
- get_step(step_id)[source]¶
Get a step by ID.
- Parameters:
step_id (str) – Step ID.
- Returns:
Step or None.
- Return type:
PlanStepNode | None
- recompute_roots()[source]¶
Recompute
rootsfrom the currentstepsdict.A root is any step that is not referenced as the
primary_inputorsecondary_inputof another step. Call this after bulk mutations (delete, hydration) where incremental root tracking is impractical.
- class veupath_chatbot.domain.strategy.session.StrategySession(site_id)[source]¶
Bases:
objectSession context for the active strategy (graph + chat).
- add_graph(graph)[source]¶
Register an existing graph in the session.
- Parameters:
graph (StrategyGraph) – Strategy graph to register.
- create_graph(name, graph_id=None)[source]¶
Create a new empty graph and register it.
- Parameters:
- Returns:
The graph.
- Return type:
- get_graph(graph_id)[source]¶
Get graph by ID (or active graph if None).
- Parameters:
graph_id (str | None) – Graph ID, or None for active graph.
- Returns:
Graph or None.
- Return type:
StrategyGraph | None
- veupath_chatbot.domain.strategy.session.hydrate_graph_from_steps_data(graph, steps_data, *, root_step_id=None, record_type=None)[source]¶
Hydrate an in-memory graph from persisted flat steps.
This is used when we have a persisted steps list (and maybe root_step_id) but no canonical plan to parse into an AST. It enables tools like list_current_steps to reflect existing UI-visible nodes.
Accepts arbitrary input; non-list values are silently ignored.
- Parameters:
graph (StrategyGraph) – Strategy graph to hydrate.
steps_data (JSONArray | object) – Flat steps list from persistence (or any value).
root_step_id (str | None) – Root step ID (default: None).
record_type (str | None) – Record type (default: None).
Research (Domain)¶
Purpose: Citation formatting and research output processing.
Citation domain types and utilities.
- class veupath_chatbot.domain.research.citations.Citation(id: str, source: Literal['web', 'europepmc', 'crossref', 'openalex', 'semanticscholar', 'pubmed', 'arxiv', 'biorxiv', 'medrxiv'], title: str, url: str | None = None, authors: list[str] | None = None, year: int | None = None, doi: str | None = None, pmid: str | None = None, snippet: str | None = None, accessed_at: str | None = None)[source]¶
Bases:
object- source: Literal['web', 'europepmc', 'crossref', 'openalex', 'semanticscholar', 'pubmed', 'arxiv', 'biorxiv', 'medrxiv']¶
- __init__(id, source, title, url=None, authors=None, year=None, doi=None, pmid=None, snippet=None, accessed_at=None)¶
- veupath_chatbot.domain.research.citations.ensure_unique_citation_tags(citations)[source]¶
Ensure all citation tags are unique by appending suffixes if needed.
- Parameters:
citations (list[JSONObject]) – Citation objects.