Dry Run Mock Generation

Dry runs validate pipeline structure without executing inference. This requires generating mock StuffContent objects that satisfy Pydantic field constraints. The DryRunFactory system produces format-compliant mock values for fields with validation rules (e.g., snake_case identifiers, PascalCase concept codes).

Why Mock Generation Matters

The structures a method declares can enforce format constraints via field_validator or model_validator:

class ComponentRecord(StructuredContent):
    component_code: str  # A validator requires snake_case
    concept_code: str  # A validator requires PascalCase

Standard mock generators (like Polyfactory) produce random strings like "uygNjiAuDMOtZEyibgHw" which fail validation. The dry run system addresses this at two levels:

1. Field-level: Generate values matching expected formats (snake_case, PascalCase, concept refs, etc.)
2. Model-level: Bypass validators using factoVry_use_construct=True

When Dry Run Mock Generation Is Used

flowchart TD
    A[pipelex validate --all] --> B[BundleValidator.validate_pipes]
    B --> C[WorkingMemoryFactory.make_mock_inputs]
    C --> D[DryRunFactory.make_dry_run_factory]
    D --> E[Polyfactory with custom providers]

    F[PipeLLM dry run] --> G[cogt leaf: dry_llm_gen_object in dry_mock.py]
    G --> H[DryRunFactory.make_dry_run_factory]
    H --> E

    I[PipeCompose dry run] --> J[StructuredContentComposer.compose]
    J --> K[model_validate with resolved values]

A dry run does not swap in a special content generator: run_mode=DRY rides CogtRunParams on every cogt assignment, and each inference leaf (llm_generate, the *_and_store image/extract leaves, search, judgment, templating) branches to its dry_* mock helper in dry_mock.py — identically whether the leaf runs inline or inside a Temporal activity.

Architecture

MockFormat Enum

Located at pipelex/cogt/content_generation/dry_run_factory.py, the MockFormat enum defines all supported mock value formats:

class MockFormat(StrEnum):
    SNAKE_CASE = "snake_case"
    PASCAL_CASE = "pascal_case"
    CONCEPT_REF = "concept_ref"
    IGNORE = "ignore"
    DICT_SNAKE_KEY_PASCAL_VALUE = "dict_snake_key_pascal_value"
    DICT_SINGLE_EXTRACT_INPUT = "dict_single_extract_input"

DryRunFactory

The factory dynamically creates a Polyfactory ModelFactory subclass with field-specific providers:

class DryRunFactory:
    @classmethod
    def generate_snake_case_code(cls) -> str:
        suffix = "".join(random.choices(string.ascii_lowercase, k=4))
        return f"mock_{suffix}"  # e.g., "mock_abcd"

    @classmethod
    def generate_pascal_case_code(cls) -> str:
        suffix = "".join(random.choices(string.ascii_lowercase, k=4))
        return f"Mock{suffix.capitalize()}"  # e.g., "MockAbcd"

    @classmethod
    def generate_concept_ref(cls) -> str:
        return f"{cls.generate_snake_case_code()}.{cls.generate_pascal_case_code()}"
        # e.g., "mock_abcd.MockCdef"

    @classmethod
    def make_dry_run_factory(
        cls,
        object_class: type[BaseModelTypeVar],
        *,
        snake_case_field_names: set[str] | None = None,
        pascal_case_field_names: set[str] | None = None,
    ) -> type[ModelFactory[BaseModelTypeVar]]: ...

Generated Mock Value Formats

A mocked image or document never carries a random string as its url, because resolve_uri takes a random string for a local path. A run with a read scope refuses to read a local path, and it checks before the dry-run branch, so a dry run on such a run would otherwise refuse its own mocks: a mocked input, or an image inside a structure a dry model call returned. An https:// URL and a data: URL are read on nobody's leave, so they are what a mock carries. The providers are keyed on the class (ImageContent, DocumentContent and their subclasses), not on a field annotation, so they apply wherever such a class is nested. They also recognise a copy of either class rebuilt from a JSON schema, by its name and its url field: a worker running a dry model call out of process gets only the output's schema, and the nested file classes it rebuilds are no subclasses of the real ones.

Declaring MockFormat on Fields

The DryRunFactory auto-detects format constraints from Pydantic Field definitions using the mock_format key in json_schema_extra:

from pydantic import Field
from pipelex.cogt.content_generation.dry_run_factory import MockFormat


class FieldDefinition(StructuredContent):
    # Snake case field
    the_field_name: str = Field(description="Field name. Must be snake_case.", json_schema_extra={"mock_format": MockFormat.SNAKE_CASE})

    # Concept reference field (domain.ConceptCode format)
    concept_ref: str | None = Field(
        default=None,
        description="For type='concept', the concept reference.",
        json_schema_extra={"mock_format": MockFormat.CONCEPT_REF},
    )

    # Field to ignore during mock generation (use default/None)
    default_value: Any | None = Field(default=None, json_schema_extra={"mock_format": MockFormat.IGNORE})

Using Field Examples for Enum-like Values

For fields that should pick from a set of valid values (like enum members or known strings), use the examples parameter. The factory's __use_examples__: True configuration makes Polyfactory randomly select from provided examples:

class Ticket(StructuredContent):
    # Refines should be one of the native concepts
    refines: str | None = Field(
        default=None,
        examples=["Text", "Image", "Document", "TextAndImages", "Number", "Page"],
    )

    # Status should be a valid TicketStatus value
    status: TicketStatus | str | None = Field(
        default=None,
        description="Where the ticket stands",
        examples=list(TicketStatus),  # Polyfactory picks randomly from these
    )

Cross-Field Dependencies with PostGenerated

For fields that depend on values generated for other fields, use Polyfactory's PostGenerated directive. The factory automatically handles the main_pipe field which must reference a key from the generated pipe dict:

# Inside DryRunFactory.make_dry_run_factory():
if "main_pipe" in object_class.model_fields and "pipe" in object_class.model_fields:
    class_attrs["main_pipe"] = PostGenerated(cls._main_pipe_from_pipe_dict)

The callback receives all previously generated values and can compute the dependent field:

@staticmethod
def _main_pipe_from_pipe_dict(_field_name: str, values: dict[str, Any]) -> str:
    pipe_dict: dict[str, Any] | None = values.get("pipe")
    if pipe_dict and len(pipe_dict) > 0:
        pipe_keys: list[str] = list(pipe_dict.keys())
        return random.choice(pipe_keys)
    # Fallback to a mock value if pipe dict is empty/not available
    return "mock_" + "".join(random.choices(string.ascii_lowercase, k=4))

Implementation Details

Mock Input Generation

When WorkingMemoryFactory.make_mock_inputs() creates mock working memory:

@classmethod
def make_mock_content(cls, typed_named_stuff_spec: TypedNamedStuffSpec) -> StuffContent:
    mock_factory = DryRunFactory.make_dry_run_factory(
        object_class=typed_named_stuff_spec.structure_class,
        snake_case_field_names=SNAKE_CASE_FIELD_NAMES,
        pascal_case_field_names=PASCAL_CASE_FIELD_NAMES,
    )
    return mock_factory.build(factory_use_construct=True)

The factory_use_construct=True flag bypasses field_validator and model_validator during object creation, preventing validation errors from random nested values.

LLM Output Generation

When the dry LLM leaf (dry_llm_gen_object in dry_mock.py) generates mock outputs, it resolves the output class exactly the way the live leaf does — via resolve_object_class — then builds via build_mock_object():

item_class = resolve_object_class(object_assignment=..., object_class=...)
return build_mock_object(item_class)

object_class is the caller's live class when there is one (the in-process case), and None when only the serialized assignment is in hand (the distributed-worker case), in which case the class is rebuilt from the assignment's JSON schema. Keeping the two run modes on one resolution matters: if only the live leaf used the real class, the mock would be built against a weaker class than the one the provider is constrained by — the mock would be less faithful than the thing it mocks.

No explicit field constraints are passed — the factory auto-detects MockFormat from field definitions. On the rebuilt-class path, json_schema_extra hints and custom validators are dropped by the schema round-trip: classes with exotic format constraints should declare examples / mock_format, otherwise re-validation against the original class raises a typed DryRunObjectFidelityError. On the live-class path those invariants are present at build time instead, so an unsatisfiable one surfaces as a typed DryRunMockBuildError naming the same remedy.

PipeCompose Resolution

StructuredContentComposer.compose() does not generate mocks. Instead, it resolves field values from working memory and validates the result:

async def compose(self) -> StuffContent:
    field_values = await self._resolve_all_fields()
    try:
        return self.output_class.model_validate(field_values)
    except ValidationError as exc:
        formatted_error = format_pydantic_validation_error(exc)
        msg = f"Cannot validate {self.output_class.__name__}: {formatted_error}"
        raise StructuredContentComposerValidationError(msg) from exc

In dry run mode, the working memory already contains properly formatted mock values (generated by WorkingMemoryFactory), so validation typically succeeds.

There is one fallback: when composition fails with a StructuredContentComposerValueError (e.g. a dotted path that mock data cannot resolve, such as a field on a description-only concept) and the run uses mock inputs, PipeCompose builds a DryRunFactory mock of the output class instead of failing. Type-incompatibility and validation errors still raise, and with real inputs (e.g. pipelex validate) unresolvable paths raise too — those are genuine bugs, not mock artifacts.

Behavior Matrix

Extending Field Constraints

Adding a New MockFormat Value

1. Add to the MockFormat enum in dry_run_factory.py:

   class MockFormat(StrEnum):
       ...
       KEBAB_CASE = "kebab_case"
   

2. Add a generator method:

   @classmethod
   def generate_kebab_case_code(cls) -> str:
       suffix = "".join(random.choices(string.ascii_lowercase, k=4))
       return f"mock-{suffix}"
   

3. Handle the format in make_dry_run_factory():

   for field_name in detected_formats[MockFormat.KEBAB_CASE]:
       if field_name in object_class.model_fields:
           class_attrs[field_name] = Use(cls.generate_kebab_case_code)
   

Using the New Format on Fields

class MyStructure(StructuredContent):
    my_kebab_field: str = Field(description="A kebab-case identifier", json_schema_extra={"mock_format": MockFormat.KEBAB_CASE})

File Reference

Next Steps

• Architecture Overview — Understand the two-layer design
• Test Profile Configuration — Configure model sets for testing