async def METHOD_NAME(flow_run, state_constructor):
    async with session.begin():
        virtual_task = await models.task_runs.create_task_run(
            session=session,
            task_run=schemas.core.TaskRun(
                flow_run_id=flow_run.id,
                task_key="a virtual task",
                dynamic_key="a virtual dynamic key",
                state=state_constructor(),
            ),
        )
        return await models.flow_runs.create_flow_run(
            session=session,
            flow_run=schemas.core.FlowRun(
                flow_id=flow.id,
                parent_task_run_id=virtual_task.id,
                state=state_constructor(),
                end_time=THE_PAST,
            ),
        )