Policies

Five composable knobs you can apply to a WorkflowSpec via .with_policy(...). Each policy conforms to a small PolicyProtocol (kind + to_wire()) so adding a sixth policy is a non-breaking change.

from convilyn_sdk import (
    WorkflowSpec,
    RetryPolicy, FailureRule, TerminalFailureRule,
    TimeoutPolicy, ToolStage,
    OutputValidationPolicy, RequiredSection, PatternCheck, StructureCheck,
    HumanReviewPolicy,
    FallbackPolicy,
)

workflow = (
    WorkflowSpec("text_analyzer")
    .with_policy(
        RetryPolicy(tool_error=3, validation_error=2),
        TimeoutPolicy(max_total_steps=40),
        OutputValidationPolicy(min_length=120),
    )
)

RetryPolicy

Per-error-class retry caps plus terminal-failure messaging.

RetryPolicy(
    tool_error=2,            # 0–10; default 2
    validation_error=1,       # 0–10; default 1
    terminal_failure=TerminalFailureRule(
        applicable_tools=["analyzer__parse"],
        message_template="Could not parse: {error_verbatim}",
    ),
    classify_failures=[
        FailureRule(
            error_code_pattern="UPSTREAM_TIMEOUT",
            log_pattern=None,
            category="transient",
            rationale_template="Upstream service timed out; retrying.",
        ),
    ],
)

TerminalFailureRule

TerminalFailureRule(
    applicable_tools=["analyzer__parse"],         # default: []
    message_template="{tool_name}: {error_verbatim}",   # default
)

{tool_name} and {error_verbatim} are the only template variables; both are filled by the platform before the message reaches the user.

FailureRule

FailureRule(
    error_code_pattern="UPSTREAM_TIMEOUT",      # required
    log_pattern=None,                            # optional
    category="transient",                        # required
    rationale_template="Upstream timeout; retrying.",   # required
)

category is one of the platform's failure taxonomy values (transient, permanent, user_input, ...). The rationale_template is what gets logged + (optionally) surfaced to support.

TimeoutPolicy

Overall step cap plus an optional staged tool pipeline.

TimeoutPolicy(
    max_total_steps=40,            # 1–200; None disables
    tool_pipeline=[
        ToolStage(name="analyzer__parse", required=True),
        ToolStage(
            name="analyzer__summarize",
            allowed_after=["analyzer__parse"],
            required=True,
        ),
    ],
)

ToolStage

ToolStage(
    name="analyzer__summarize",       # required
    allowed_after=["analyzer__parse"], # default: []
    required=True,                     # default: True
)

OutputValidationPolicy

What the final artefact must contain. Every field is independently optional — a partial policy only constrains the listed dimensions and leaves the rest to default prose-driven behaviour.

OutputValidationPolicy(
    description="The summary must include both an abstract and a bullet list.",
    required_sections=[
        RequiredSection(name="abstract", synonyms=["summary", "tl;dr"]),
        RequiredSection(name="bullets"),
    ],
    anti_patterns=["TODO", "FIXME", "lorem ipsum"],
    min_length=120,
    pattern_checks=[
        PatternCheck(
            type="must_contain",
            patterns=[r"\babstract\b"],
            reason="The artefact must lead with the abstract.",
            min_matches=1,
        ),
    ],
    structure_checks=[
        StructureCheck(type="bullet_list", count=3),
    ],
)

RequiredSection

RequiredSection(
    name="abstract",                         # required
    synonyms=["summary", "tl;dr"],            # default: []
)

PatternCheck

PatternCheck(
    type="must_contain",                # QualityCheckType — must_contain / must_not_contain / ...
    patterns=[r"\babstract\b"],
    reason="Lead-with-abstract rule from the style guide.",
    min_matches=1,                       # >=1
)

StructureCheck

StructureCheck(type="bullet_list", count=3)

type is one of the platform's StructuralCheckType values (bullet_list, numbered_list, heading, ...). count is the minimum (>=0).

HumanReviewPolicy

How slots are emitted when the agent pauses for user input.

HumanReviewPolicy(
    files_only=False,                                   # restrict slots to file uploads
    allowed_slot_types=["text", "choice"],               # per-workflow allowlist; empty = all
    first_question_format="natural_language",            # pins the very first prompt's shape
)

FallbackPolicy

Clarify / infer / stop vocabularies plus the no-tool-result action.

FallbackPolicy(
    must_clarify_when=["missing_required_input"],
    may_infer_when=["context_partial"],
    must_stop_when=["irrecoverable_state"],
    on_no_tool_result="fail_safe",      # "fail_safe" | "retry" | "continue"
)

The three condition lists are disjoint by design; populating one list does not affect the others.

Wire-shape Config siblings

Every policy above compiles into a Pydantic config sibling that ships in convilyn_sdk.workflow_policies. Authors do not normally need to touch them — with_policy(...) does the translation — but they are exported for advanced cases (custom validators, manual wire-shape inspection):

from convilyn_sdk import (
    FailureRuleConfig, FallbackPolicyConfig, GoalCriteriaConfig,
    QaPolicyConfig, QualityCheckConfig, RetryPolicyConfig,
    RoutingPolicyConfig, SectionConfig, SlotPolicyConfig,
    StructuralCheckConfig, TaskPolicyConfig,
    TerminalFailurePolicyConfig, ToolStageConfig,
)

The Config models set extra="allow", so a server-side schema addition does not force an SDK bump — authors round-trip the new field via compile() → model_dump() without the SDK knowing about it.

Compositional ordering

with_policy is variadic and order-independent:

workflow.with_policy(retry_policy, timeout_policy, output_policy)

is identical to

workflow.with_policy(timeout_policy).with_policy(output_policy).with_policy(retry_policy)

Policies of the same kind are last-write-wins; the builder rebuilds the wire dict from the latest values on each compile().