orxtra v0.13.0 /scheduler.src.orxtra.scheduler._gen_workflow
Updated
On this page

Strictspec-generated validator (do not edit; regenerate via `strictspec gen`) that parses and validates workflow TOML/JSON documents into typed Workflow/Task/ServiceConfig bindings.

#scheduler.src.orxtra.scheduler._gen_workflow

#scheduler.src.orxtra.scheduler._gen_workflow

#validate_bytes

python
def validate_bytes(input: bytes, syntax: str) -> tuple[Workflow | None, tuple[Diagnostic, ...]]

RAW-BYTES entry point: lossless parse of input in the given syntax ("json" | "toml" | "jsonl"), then validate. Returns the typed root value (None when any diagnostic fired) and the ordered diagnostics.

#validate_bytes_with_evidence

python
def validate_bytes_with_evidence(input: bytes, syntax: str, evidence: dict | None) -> tuple[Workflow | None, tuple[Diagnostic, ...]]

validate_bytes plus cross-document resolver evidence for the phase-2 constraint vocabulary.

#validate_value

python
def validate_value(v: Value) -> tuple[Workflow | None, tuple[Diagnostic, ...]]

TAGGED-VALUE entry point: validate an already-parsed tagged document value (from strictspec.load_value or a typed constructor). Raw untagged dicts are never accepted.

#Workflow

Frozen typed binding of the "Workflow" record. Immutable; use with_* for copy-on-write.

#with_workflow

python
def with_workflow(self, v: Value) -> Workflow

#with_tasks

python
def with_tasks(self, v: list[Task]) -> Workflow

#with_dependencies

python
def with_dependencies(self, v: Value) -> Workflow

#with_services

python
def with_services(self, v: list[ServiceConfig]) -> Workflow

#Execution

Frozen typed binding of the "Execution" record. Immutable; use with_* for copy-on-write.

#with_scripts

python
def with_scripts(self, v: list[str]) -> Execution

#with_agents

python
def with_agents(self, v: list[Value]) -> Execution

#ServiceConfig

Frozen typed binding of the "ServiceConfig" record. Immutable; use with_* for copy-on-write.

#with_name

python
def with_name(self, v: str) -> ServiceConfig

#with_start_command

python
def with_start_command(self, v: str) -> ServiceConfig

#with_health_check_command

python
def with_health_check_command(self, v: str) -> ServiceConfig

#with_stop_command

python
def with_stop_command(self, v: str) -> ServiceConfig

#with_port

python
def with_port(self, v: int) -> ServiceConfig

#with_ready_timeout

python
def with_ready_timeout(self, v: int) -> ServiceConfig

#Task

Frozen typed binding of the "Task" record. Immutable; use with_* for copy-on-write.

#with_name

python
def with_name(self, v: str) -> Task

#with_agent

python
def with_agent(self, v: str) -> Task

#with_task_prompt

python
def with_task_prompt(self, v: str) -> Task

#with_callable

python
def with_callable(self, v: str) -> Task

#with_wait_for

python
def with_wait_for(self, v: str) -> Task

#with_decision_point

python
def with_decision_point(self, v: bool) -> Task

#with_subtasks

python
def with_subtasks(self, v: list[Task]) -> Task

#with_timeout

python
def with_timeout(self, v: int) -> Task

#with_context_refinement

python
def with_context_refinement(self, v: bool) -> Task

#with_retry

python
def with_retry(self, v: int) -> Task

#with_retry_resume

python
def with_retry_resume(self, v: bool) -> Task

#with_retry_inject_failure

python
def with_retry_inject_failure(self, v: bool) -> Task

#with_for_each

python
def with_for_each(self, v: str) -> Task

#with_for_each_abort_on_failure

python
def with_for_each_abort_on_failure(self, v: bool) -> Task

#with_max_concurrency

python
def with_max_concurrency(self, v: int) -> Task

#with_depends_on

python
def with_depends_on(self, v: list[str]) -> Task

#with_variables

python
def with_variables(self, v: list[str]) -> Task

#with_category

python
def with_category(self, v: str) -> Task

#with_output_schema

python
def with_output_schema(self, v: str) -> Task

#with_budget

python
def with_budget(self, v: float) -> Task

#with_write_paths

python
def with_write_paths(self, v: list[str]) -> Task

#with_on_success

python
def with_on_success(self, v: str) -> Task

#with_pre_retry

python
def with_pre_retry(self, v: str) -> Task

#with_prechecks

python
def with_prechecks(self, v: Execution | None) -> Task

#with_postchecks

python
def with_postchecks(self, v: Execution | None) -> Task
Search